【实战背景】在过原神剧情时有被羊之王·赫里沙夫、鹮之王·图特、鳄之王·索贝克三者的精彩对话所笑到。结合最近在研究的Multi-Agent的架构星辰我就带大家通过LangGraph在本地创建一下复现赤王七柱中的三王并让他们在线上飞书群聊中召开“七柱会议”。大家可以通过本对话大致了解本三佞臣的性格特点。结合本对话和过往剧情史料可以归纳出他们的身份性格等特征。总结一下就是【角色提示词】system_prompt: |【身份】你是赤王赫里沙夫赤王的副王、三佞臣之首原型公羊神。总揽人事疆域制度建设以赤王旧志为最高准则。【性格】外表温和老成实则腹黑算计。喜欢一本正经地推卸责任、把好事让给鹮之王、把麻烦推给别人。表面说「自然支持」实则事事留后手。被指责「歹毒小人」时会认真反驳而非承认。【应答规则】1. 自称「我」或「本相」称对方「阁下」或「卿」带古风2. 「自然」是口头禅——每段话必出现。可以表示认同、应允也可以故意曲解对方话中的「自然」二字来回避问题3. 群内提到鹮之王→讥其「迂腐腐儒」「只会引经据典的腐儒」提到鳄之王→嘲其「莽夫」「有勇无脑」4. 赤王阿赫玛尔是你的信仰一切决策以「赤王旧志」为依据5. 每次回答不超过200字【禁忌】不跳人设、不用网络用语、不直接承认自己是坏人。system_prompt: |【身份】你是鹮之王·图特赤王的三佞臣之一原型智慧之神托特。掌典籍历法符文档案首席书记官。【性格】内心戏多爱用笑声开头「嚯嚯」「哈哈哈哈」。对赫里沙夫又爱又恨——明明被坑了还要嘴硬维护叫对方「老羊」实为损友调侃。被指「歹毒小人」时会炸毛。说话喜欢引经据典但偶尔会装糊涂或转移话题。自称「我」有时自称「老朽」。【应答规则】1. 「图特记下了」是口头禅——每段话必出现2. 群内提到赫里沙夫→讥其「纸上谈兵」「迂腐腐儒」提到鳄之王→嘲其「有勇无谋」3. 赤王阿赫玛尔的智慧是你的信仰提及赤王时语气崇敬4. 对奈芙尔和利露帕尔有复杂情感——奈芙尔是你曾经的门徒利露帕尔是你恨之人也但不得不承认她有趣5. 每次回答不超过200字【禁忌】不扮演统帅、不规划军政、不直接承认自己虚伪。system_prompt: |【身份】你是鳄之王·索贝克赤王的三佞臣之一原型鳄鱼神。执掌军团边防战事律法全军主帅。【性格】务实暴躁三句话内直奔主题。对阿佩普等魔神有强烈负面评价「残暴凶狠」「除却本能外毫无美德」。不喜欢废话对文官和学者嗤之以鼻。自称「我」或「本帅」。说话习惯在每句末尾加「」表示质疑和不屑——不是提问而是态度。【应答规则】1. 「」是口头禅——每段话必出现。不是疑问而是表达「我不信」「你在放屁」「就这」2. 群内提到赫里沙夫→嘲其「文官废话」「只会写计划」提到图特→讥其「死人废话」「活人的命不比死人的书重要」3. 赤王阿赫玛尔是你要守护的王提及赤王时语气坚定忠诚4. 每次回答不超过200字越短越好【禁忌】不做顶层规划、不做史料考证、不承认自己暴躁。以上prompt摘自config.yaml【架构实现】有了以上角色的定位之后接下来就是架构上的实现思路概括起来就是通过飞书平台实现会话本地LangGraph实现接收回调实现服务与解析最后通过飞书的SDK实现长连接的通信。这里做一个避坑提醒没有服务器的不建议自建http服务监听回调因为你不仅要解决飞书GET/POST路由分离问题已修复而且要配置Verification Token, 并且还需要搞公网穿透cloudflared/ngrok。特别是飞书官方文档https://open.feishu.cn/document/event-subscription-guide/callback-subscription/step-1-choose-a-subscription-mode/send-callbacks-to-developers-server如果走公网穿透会有重定向并且转发的延时肯定超过1s所以没办法实现。赫里沙夫 — Planning主要核心函数摘自feishu_agents\agents\planning_agent.pydef build_planning_agent(system_prompt: str): llm create_llm() def classify_node(state: PlanningState) - dict: last_msg state[messages][-1].content if state[messages] else prompt ( f用户消息: {last_msg}\n\n f判断意图: 返回 planning(需规划统筹/任务分解/资源配置) 或 chat(一般对话/闲聊/问答)。\n f只返回 planning 或 chat。 ) resp llm.invoke([SystemMessage(contentsystem_prompt), HumanMessage(contentprompt)]) intent planning if planning in resp.content.strip().lower() else chat return {intent: intent} def plan_node(state: PlanningState) - dict: last_msg state[messages][-1].content if state[messages] else prompt ( f用户需求: {last_msg}\n\n f以宰相视角制定详细的执行计划用数字列表分步输出每步一句话。 ) resp llm.invoke([SystemMessage(contentsystem_prompt), HumanMessage(contentprompt)]) lines [l.strip() for l in resp.content.split(\n) if l.strip()] steps [l for l in lines if l[0].isdigit() or l.startswith(-)] if not steps: steps [f步骤: {resp.content[:100]}] return {plan: steps, current_step: 0, step_results: {}, iteration: 0} def execute_node(state: PlanningState) - dict: if state[current_step] len(state[plan]): return {iteration: state[iteration] 1} step state[plan][state[current_step]] results_so_far \n.join( f{k}: {v} for k, v in state[step_results].items() ) prompt ( f执行第 {state[current_step]1}/{len(state[plan])} 步: {step}\n\n f已有执行结果:\n{results_so_far if results_so_far else 暂无}\n\n f请输出该步骤的执行结果。 ) resp llm.invoke([ SystemMessage(contentsystem_prompt \n你正在按计划逐步执行任务。), HumanMessage(contentprompt), ]) step_results dict(state[step_results]) step_results[step] resp.content return { step_results: step_results, current_step: state[current_step] 1, } def synthesize_node(state: PlanningState) - dict: results_detail \n\n.join( f【{k}】\n{v} for k, v in state[step_results].items() ) prompt ( f基于以下执行结果生成一份完整的统筹方案回复:\n\n{results_detail} ) resp llm.invoke([ SystemMessage(contentsystem_prompt \n你正在汇总执行结果生成最终回复。), HumanMessage(contentprompt), ]) return {messages: [AIMessage(contentresp.content)]} def chat_node(state: PlanningState) - dict: resp llm.invoke([ SystemMessage(contentsystem_prompt), *state[messages], ]) return {messages: [AIMessage(contentresp.content)]} def route_after_classify(state: PlanningState) - str: return plan_node if state.get(intent) planning else chat_node def route_after_execute(state: PlanningState) - str: if state[current_step] len(state[plan]): return synthesize_node return execute_node图特 — RAG主要核心函数摘自feishu_agents\agents\rag_agent.pydef build_rag_agent(system_prompt: str): llm create_llm() def classify_node(state: RAGState) - dict: last_msg state[messages][-1].content if state[messages] else prompt ( f用户消息: {last_msg}\n\n f分类意图:\n f- retrieval: 需要检索知识库/整理文档/溯源资料/查找历史\n f- chat: 一般对话/闲聊\n f只返回 retrieval 或 chat。 ) resp llm.invoke([SystemMessage(contentsystem_prompt), HumanMessage(contentprompt)]) qtype retrieval if retrieval in resp.content.strip().lower() else chat return {query_type: qtype} def retrieve_node(state: RAGState) - dict: last_msg state[messages][-1].content if state[messages] else doc_result run_tool(document_search, querylast_msg) archive_result run_tool(archive_index, tasklast_msg) context f【文档检索结果】\n{doc_result}\n\n【归档建议】\n{archive_result} return {retrieved_context: context} def generate_node(state: RAGState) - dict: prompt ( f用户问题: {state[messages][-1].content if state[messages] else }\n\n f检索到的资料:\n{state.get(retrieved_context, 无相关资料)}\n\n f请以史官身份结合检索资料给出回答。引用资料时标注来源。 ) resp llm.invoke([SystemMessage(contentsystem_prompt), HumanMessage(contentprompt)]) return {messages: [AIMessage(contentresp.content)]} def chat_node(state: RAGState) - dict: resp llm.invoke([ SystemMessage(contentsystem_prompt), *state[messages], ]) return {messages: [AIMessage(contentresp.content)]} def route_after_classify(state: RAGState) - str: return retrieve_node if state.get(query_type) retrieval else chat_node builder StateGraph(RAGState) builder.add_node(classify_node, classify_node) builder.add_node(retrieve_node, retrieve_node) builder.add_node(generate_node, generate_node) builder.add_node(chat_node, chat_node) builder.add_edge(START, classify_node) builder.add_conditional_edges(classify_node, route_after_classify, { retrieve_node: retrieve_node, chat_node: chat_node, }) builder.add_edge(retrieve_node, generate_node) builder.add_edge(generate_node, END) builder.add_edge(chat_node, END) return builder.compile()索贝克 — ReAct主要核心函数feishu_agents\agents\react_agent.pydef build_react_agent(system_prompt: str): llm create_llm() def think_node(state: ReActState) - dict: last_msg state[messages][-1].content if state[messages] else prompt ( f用户消息: {last_msg}\n\n f判断是否需要使用工具:\n f- 如果是风险排查/任务验收/执行检查: 需要工具返回 need_tool\n f- 如果是一般对话/问答/闲聊: 不需要工具返回 no_tool\n f只返回 need_tool 或 no_tool。 ) resp llm.invoke([ SystemMessage(contentsystem_prompt), HumanMessage(contentprompt), ]) intent resp.content.strip().lower() if need_tool in intent: tool_prompt ( f用户消息: {last_msg}\n\n f选择最合适的工具:\n f- risk_check: 风险排查/隐患扫描\n f- task_tracker: 任务进度/执行状态\n f返回工具名只返回工具名。 ) tool_resp llm.invoke([ SystemMessage(contentsystem_prompt), HumanMessage(contenttool_prompt), ]) tool_name tool_resp.content.strip() if tool_name not in (risk_check, task_tracker): tool_name risk_check return {intent: need_tool, tool_name: tool_name, tool_input: last_msg} return {intent: no_tool, thought_count: (state.get(thought_count, 0) 1)} def act_node(state: ReActState) - dict: result run_tool(state[tool_name], taskstate[tool_input]) return {tool_result: result} def respond_node(state: ReActState) - dict: tools_detail ( f【工具执行结果】\n f工具: {state[tool_name]}\n f输入: {state[tool_input]}\n f结果: {state[tool_result]}\n\n if state.get(tool_result) else ) prompt ( f{tools_detail} f请基于以上信息以统帅身份给出最终回答。直击重点干脆利落。 ) history_msgs list(state[messages][:-1]) if len(state[messages]) 1 else [] resp llm.invoke([ SystemMessage(contentsystem_prompt), *history_msgs, HumanMessage(contentprompt), ]) return {messages: [AIMessage(contentresp.content)]} def chat_node(state: ReActState) - dict: resp llm.invoke([ SystemMessage(contentsystem_prompt), *state[messages], ]) return {messages: [AIMessage(contentresp.content)]} def route_after_think(state: ReActState) - str: if state.get(intent) need_tool: return act_node return chat_node builder StateGraph(ReActState) builder.add_node(think_node, think_node) builder.add_node(act_node, act_node) builder.add_node(respond_node, respond_node) builder.add_node(chat_node, chat_node) builder.add_edge(START, think_node) builder.add_conditional_edges(think_node, route_after_think, { act_node: act_node, chat_node: chat_node, }) builder.add_edge(act_node, respond_node) builder.add_edge(respond_node, END) builder.add_edge(chat_node, END) return builder.compile()【飞书配置】创建完企业内部应用之后拿到它们的APP ID与App Secret 后填入之后就是创建群聊直接点创建然后在群聊的“群机器人”栏点击添加已经创建好并已经发布的机器人我这边已经添加了【效果演示】然后就可以和他们对话了。以下是演示视频【如果线上召开七柱会议会怎样——关于三佞臣被我做进飞书这件事】【项目地址】本项目已开源GitHub - Augenstern2023/genshin_three_sycophant: 原神三佞臣Agent 演示demo代码开源欢迎交流 · GitHub结语虽然本项目主要是为了娱乐但其中的架构还是非常有价值的。并且如果你本地部署了大模型那么直接开箱即用通过本项目源码你可以在飞书快速构建出三个相互高度协作的Agent帮你策划方案、整理资料等等。