多Agent 飞书对话实战——以还原【原神】中三佞臣为例
2026/6/8 8:12:58 网站建设 项目流程

【实战背景】

在过原神剧情时,有被羊之王·赫里沙夫、鹮之王·图特、鳄之王·索贝克,三者的精彩对话所笑到。结合最近在研究的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.py)

def 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(content=system_prompt), HumanMessage(content=prompt)]) 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(content=system_prompt), HumanMessage(content=prompt)]) 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(content=system_prompt + "\n你正在按计划逐步执行任务。"), HumanMessage(content=prompt), ]) 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(content=system_prompt + "\n你正在汇总执行结果,生成最终回复。"), HumanMessage(content=prompt), ]) return {"messages": [AIMessage(content=resp.content)]} def chat_node(state: PlanningState) -> dict: resp = llm.invoke([ SystemMessage(content=system_prompt), *state["messages"], ]) return {"messages": [AIMessage(content=resp.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.py)

def 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(content=system_prompt), HumanMessage(content=prompt)]) 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", query=last_msg) archive_result = run_tool("archive_index", task=last_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(content=system_prompt), HumanMessage(content=prompt)]) return {"messages": [AIMessage(content=resp.content)]} def chat_node(state: RAGState) -> dict: resp = llm.invoke([ SystemMessage(content=system_prompt), *state["messages"], ]) return {"messages": [AIMessage(content=resp.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.py)

def 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(content=system_prompt), HumanMessage(content=prompt), ]) 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(content=system_prompt), HumanMessage(content=tool_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"], task=state["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(content=system_prompt), *history_msgs, HumanMessage(content=prompt), ]) return {"messages": [AIMessage(content=resp.content)]} def chat_node(state: ReActState) -> dict: resp = llm.invoke([ SystemMessage(content=system_prompt), *state["messages"], ]) return {"messages": [AIMessage(content=resp.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,帮你策划方案、整理资料等等。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询