基于langgraph的智能问答工作流

做一个人入口用户问啥答啥但内部自动分流。通用问题走通用问答 Agent代码相关走代码处理 Agent。最关键的是——要支持多轮对话用户能追问、补充代码、深入聊。最后折腾出来的方案用 LangGraph 搭路由 Agent 组合的工作流接 MiniMax API配合 LangSmith 做调用链追踪State 里用add_messagesreducer 自动累积对话历史。开工前的准备初始化项目mkdir demos cd demos uv init装依赖uv add langchain-openai langgraph python-dotenv uv add langchain-core langchain-communitypyproject.toml最终内容[project] name demos version 0.1.0 requires-python 3.12 dependencies [ langchain-openai1.1.12, langgraph1.0.0, python-dotenv1.2.2, ]配置 API Key根目录新建.envMINIMAX_API_KEY 你的MiniMax密钥 SMITH_API_KEY 你的LangSmith密钥目录结构demos/ ├── agents/ # Agent 定义 │ ├── __init__.py │ ├── code_agent.py # 代码处理 Agent │ └── prompt_agent.py # 通用问答 Agent ├── core/ # 核心工具 │ ├── __init__.py │ ├── llm.py # LLM 初始化和调用封装 │ └── tracing.py # LangSmith 配置 ├── tools/ # 工具集 │ ├── __init__.py │ ├── math_tools.py # 数学计算工具 │ └── search_tools.py # 搜索工具 ├── workflow/ # 工作流 │ ├── nodes/ # 节点实现 │ ├── graph/ # 图定义 │ ├── routes/ # 路由逻辑 │ ├── states/ # 状态定义 │ └── simple_assistant/ └── run_workflow.py # 入口脚本核心模块LLM 封装core/llm.py统一管理模型调用import os from langchain_openai import ChatOpenAI def build_llm() - ChatOpenAI: api_key os.getenv(MINIMAX_API_KEY) return ChatOpenAI( modelMiniMax-M2.7, base_urlhttps://api.minimaxi.com/v1, api_keyapi_key, temperature0.7, max_tokens1000, timeout60, )LangSmith 配置core/tracing.py处理调用链追踪import os from dotenv import load_dotenv def configure_langsmith(project_name: str) - str: load_dotenv() api_key os.getenv(LANGSMITH_API_KEY) or os.getenv(SMITH_API_KEY) if api_key: os.environ[LANGSMITH_API_KEY] api_key os.environ[LANGSMITH_TRACING] true os.environ[LANGSMITH_PROJECT] project_name return project_name def build_run_config(run_name: str, tagsNone, metadataNone): return { run_name: run_name, tags: list(tags or []), metadata: dict(metadata or {}), } def extend_run_config(config, run_nameNone, tagsNone, metadataNone): from langchain_core.runnables.config import merge_configs extra { tags: list(tags or []), metadata: dict(metadata or {}), } if run_name: extra[run_name] run_name return merge_configs(config, extra)Agent 实现通用问答 Agentagents/prompt_agent.pyPromptAgent类提供reply()方法输入问题字符串返回 dict。核心能力是根据用户输入自动判断是否需要调用计算或搜索工具。代码处理 Agentagents/code_agent.pyCodeAgent类提供reply()和debug_reply()方法分别处理普通代码问答和带报错信息的代码调试场景。工作流设计State 定义支持多轮对话的关键workflow/states/simple_assistant_state.pyfrom typing import Annotated, Literal, TypedDict from langgraph.graph import add_messages class SimpleAssistantState(TypedDict): messages: Annotated[list, add_messages] code: str error_message: str expected_behavior: str language: str intent: Literal[prompt, code] route_reason: str agent_name: str scenario: str tool_route: str重点在messages: Annotated[list, add_messages]——这行让新消息自动追加到列表而不是覆盖。add_messages是 LangGraph 内置的 reducer 函数专门处理消息合并。路由逻辑workflow/routes/simple_assistant_routes.pyfrom typing import Literal from langchain_core.messages import BaseMessage from workflow.states.simple_assistant_state import SimpleAssistantState CODE_HINT_KEYWORDS ( python, java, javascript, 代码, 函数, 类, 报错, 异常, 错误, 修复, debug, bug, traceback, stack trace, review, ) def _get_latest_user_message(messages: list[BaseMessage]) - str: for msg in reversed(messages): if hasattr(msg, type) and msg.type human: return msg.content return def detect_intent(state: SimpleAssistantState) - tuple[Literal[prompt, code], str]: if state.get(code) or state.get(error_message): return code, state中包含code或error_message messages state.get(messages, []) user_input _get_latest_user_message(messages).strip().lower() if any(keyword in user_input for keyword in CODE_HINT_KEYWORDS): return code, 命中了编程/报错关键词 return prompt, 默认走通用问答agent def route_after_router(state: SimpleAssistantState) - Literal[prompt_agent_node, code_agent_node]: if state.get(intent) code: return code_agent_node return prompt_agent_node路由逻辑先看 state 里有没有 code 或 error_message有的话直接走 code_agent没有的话从 messages 里提取最新用户输入命中关键词就走 code_agent否则走 prompt_agent。节点实现workflow/nodes/simple_assistant_nodes.pyfrom langchain_core.messages import AIMessage, HumanMessage from langchain_core.runnables import RunnableConfig from agents.code_agent import code_agent from agents.prompt_agent import prompt_agent from core.tracing import extend_run_config from workflow.routes.simple_assistant_routes import detect_intent from workflow.states.simple_assistant_state import SimpleAssistantState def router_node(state: SimpleAssistantState) - dict: intent, route_reason detect_intent(state) return {intent: intent, route_reason: route_reason} def prompt_agent_node(state: SimpleAssistantState, *, config: RunnableConfig) - dict: messages state.get(messages, []) last_human_msg for msg in reversed(messages): if isinstance(msg, HumanMessage): last_human_msg msg.content break agent_config extend_run_config( config, run_nameprompt_agent_node, tags[node, prompt_agent_node], ) try: result prompt_agent.reply(last_human_msg, configagent_config) return { messages: [AIMessage(contentresult[answer])], agent_name: result[agent_name], answer: result[answer], thinking: result[thinking], tool_route: result[tool_route], scenario: prompt_answer, } except Exception as exc: return { messages: [AIMessage(contentfprompt_agent 调用失败{exc})], agent_name: prompt_agent, answer: fprompt_agent 调用失败{exc}, thinking: , tool_route: general, scenario: prompt_error, } def code_agent_node(state: SimpleAssistantState, *, config: RunnableConfig) - dict: messages state.get(messages, []) last_human_msg for msg in reversed(messages): if isinstance(msg, HumanMessage): last_human_msg msg.content break agent_config extend_run_config( config, run_namecode_agent_node, tags[node, code_agent_node], ) try: if state.get(code) or state.get(error_message): result code_agent.debug_reply( tasklast_human_msg, codestate.get(code, ), error_messagestate.get(error_message, ), expected_behaviorstate.get(expected_behavior, ), languagestate.get(language, Python), configagent_config, ) else: result code_agent.reply(last_human_msg, configagent_config) except Exception as exc: return { messages: [AIMessage(contentfcode_agent 调用失败{exc})], agent_name: code_agent, answer: fcode_agent 调用失败{exc}, thinking: , scenario: code_error, tool_route: general, } return { messages: [AIMessage(contentresult[answer])], agent_name: result[agent_name], answer: result[answer], thinking: result[thinking], scenario: result[scenario], tool_route: state.get(tool_route, general), }节点从 state 取 messages 列表拿到最后一条 HumanMessage 传给 Agent 处理。返回值里的messages字段会被 LangGraph 自动合并到历史里。图构建workflow/graph/simple_assistant_graph.pyfrom typing import Any from langgraph.graph import END, START, StateGraph from workflow.nodes.simple_assistant_nodes import code_agent_node, prompt_agent_node, router_node from workflow.routes.simple_assistant_routes import route_after_router from workflow.states.simple_assistant_state import SimpleAssistantState from core.tracing import build_run_config, configure_langsmith DEFAULT_WORKFLOW_PROJECT demos-simple-assistant def build_simple_assistant_graph(): graph StateGraph(SimpleAssistantState) graph.add_node(router_node, router_node, metadata{step: routing}) graph.add_node(prompt_agent_node, prompt_agent_node, metadata{step: agent, agent: prompt_agent}) graph.add_node(code_agent_node, code_agent_node, metadata{step: agent, agent: code_agent}) graph.add_edge(START, router_node) graph.add_conditional_edges(router_node, route_after_router) graph.add_edge(prompt_agent_node, END) graph.add_edge(code_agent_node, END) return graph.compile(namesimple_assistant_graph) app build_simple_assistant_graph() def run_simple_assistant( messages: list, *, code: str , error_message: str , expected_behavior: str , language: str Python, project_name: str DEFAULT_WORKFLOW_PROJECT, ) - dict[str, Any]: langsmith_project configure_langsmith(project_name) state: SimpleAssistantState { messages: messages, code: code, error_message: error_message, expected_behavior: expected_behavior, language: language, } run_config build_run_config( run_namesimple_assistant_run, tags[workflow, simple_assistant], metadata{ workflow: simple_assistant, langsmith_project: langsmith_project, has_code_context: bool(code or error_message), }, ) return app.invoke(state, configrun_config)流程START → router_node 做分流 → 根据 intent 走 prompt_agent_node 或 code_agent_node → END。入口脚本run_workflow.pyimport argparse from langchain_core.messages import HumanMessage from workflow.graph import app, run_simple_assistant def print_graph(): print(\n * 60) print(Workflow Graph:) print( * 60) graph app.get_graph() print(graph.draw_mermaid()) print( * 60 \n) def main(): parser argparse.ArgumentParser(descriptionSimple Assistant Workflow) parser.add_argument(--show-graph, actionstore_true, helpShow workflow graph) parser.add_argument(--user-input, typestr, helpUser input for the workflow) parser.add_argument(--code, typestr, default, helpCode snippet (optional)) parser.add_argument(--error, typestr, default, helpError message (optional)) parser.add_argument(--expected, typestr, default, helpExpected behavior (optional)) parser.add_argument(--language, typestr, defaultPython, helpProgramming language) args parser.parse_args() if args.show_graph: print_graph() return messages [] if args.user_input: messages.append(HumanMessage(contentargs.user_input)) result run_simple_assistant( messagesmessages, codeargs.code, error_messageargs.error, expected_behaviorargs.expected, languageargs.language, ) messages result.get(messages, []) print(\n * 60) print(Answer:) print( * 60) if messages and hasattr(messages[-1], content): print(messages[-1].content) print( * 60) else: print(Simple Assistant Workflow CLI (多轮对话模式)) print( * 60) print(输入 exit 或 quit 退出对话) print() while True: user_input input(你: ).strip() if not user_input: continue if user_input.lower() in (exit, quit, q): print(再见!) break messages.append(HumanMessage(contentuser_input)) print(思考中...\n) result run_simple_assistant( messagesmessages, code, error_message, expected_behavior, languagePython, ) messages result.get(messages, []) if messages and hasattr(messages[-1], content): print(f\n助手: {messages[-1].content}\n) if __name__ __main__: main()支持两种模式--user-input单轮问答不带参数进入多轮对话循环怎么跑看工作流图uv run python run_workflow.py --show-graph输出 Mermaid 格式的图结构复制到 mermaid.live 直接渲染。单轮问答uv run python run_workflow.py --user-input 你好介绍一下自己多轮对话uv run python run_workflow.py示例Simple Assistant Workflow CLI (多轮对话模式) 输入 exit 或 quit 退出对话 你: 帮我解释一下什么是闭包 助手: 闭包是... 你: 能举个Python的例子吗 助手: 当然这里是一个例子... 你: exit 再见!多轮对话怎么实现的核心在 State 里的messages: Annotated[list, add_messages]。add_messages是 LangGraph 内置的 reducer 函数作用是告诉 LangGraph同一字段多次更新时怎么处理合并。默认行为是覆盖但指定了 reducer 后LangGraph 会把新消息 append 到列表而不是整个列表替换掉。所以流程是这样的用户输入你好→ messages [HumanMessage(content你好)]节点返回{messages: [AIMessage(content你好有什么可以帮你的)]}LangGraph 自动合并 → messages [HumanMessage(...), AIMessage(...)]用户追问你能做什么→ messages [HumanMessage(...), AIMessage(...), HumanMessage(content你能做什么)]节点看到的是完整历史可以基于之前的上下文回答关键点节点返回值必须包含messages字段LangGraph 才会触发合并逻辑。项目结构最终版demos/ ├── agents/ │ ├── __init__.py │ ├── code_agent.py │ └── prompt_agent.py ├── core/ │ ├── __init__.py │ ├── llm.py │ └── tracing.py ├── tools/ │ ├── __init__.py │ ├── math_tools.py │ └── search_tools.py ├── workflow/ │ ├── graph/ │ │ ├── __init__.py │ │ └── simple_assistant_graph.py │ ├── nodes/ │ │ ├── __init__.py │ │ └── simple_assistant_nodes.py │ ├── routes/ │ │ ├── __init__.py │ │ └── simple_assistant_routes.py │ ├── states/ │ │ ├── __init__.py │ │ └── simple_assistant_state.py │ └── simple_assistant/ │ └── __init__.py ├── .env ├── .gitignore ├── pyproject.toml └── run_workflow.py