如何在项目中落地Agent？从选型到上线的完整指南 1. 引言大语言模型LLM的能力已经毋庸置疑但如何让它在真实的业务场景中稳定、可靠地完成任务而不是仅仅做一个“聊天机器人”答案就是Agent智能体。Agent 的核心在于让 LLM 具备感知、规划、行动和记忆的能力从而自主地完成复杂任务。然而从概念到生产落地中间隔着“选型、架构、开发、调试、上线”五座大山。本文将带你手把手走完这一完整流程从技术选型开始到最终部署一个能稳定运行的 Agent 服务并提供可直接运行的代码示例。2. 第一步技术选型——选择合适的 Agent 框架在开始编码前我们需要选择一个合适的框架。目前主流的 Agent 框架有 LangChain、AutoGPT、CrewAI 等。对于企业级项目落地LangChain和LangGraph是目前最成熟、生态最完善的选择。2.1 为什么选择 LangChain LangGraphLangChain提供了丰富的 LLM 封装、工具集成Tool、记忆管理Memory和链式调用Chain是构建 Agent 的基础。LangGraph在 LangChain 之上允许你用有向图Graph的方式定义 Agent 的执行流程支持循环、分支和状态管理非常适合构建复杂的、可预测的 Agent 行为。2.2 环境准备首先安装必要的依赖库pipinstalllangchain langchain-openai langgraph langchain-community注意请确保你已配置好 OpenAI API Key 或其他兼容的 LLM API如通义千问、DeepSeek 等。本教程以 OpenAI 为例。3. 第二步架构设计——定义 Agent 的核心能力一个标准的 Agent 架构包含以下几个核心组件LLM大脑负责推理、规划和决策。Tools手脚Agent 可以调用的外部能力如搜索引擎、数据库查询、API 调用、代码执行器等。Memory记忆记录对话历史和中间状态让 Agent 具备上下文理解能力。Planning规划将复杂任务分解为可执行的步骤。3.1 定义工具Tools工具是 Agent 与外部世界交互的桥梁。我们以两个常见场景为例获取天气和执行数学计算。# tools.pyfromlangchain_core.toolsimporttoolimportrequeststooldefget_weather(city:str)-str:获取指定城市的当前天气情况。# 这里使用一个模拟的天气 API实际项目中请替换为真实 API# 例如https://api.openweathermap.org/weather_data{北京:晴朗25°C,上海:多云28°C,深圳:阵雨30°C,}returnweather_data.get(city,f抱歉没有找到{city}的天气信息。)tooldefcalculate(expression:str)-str:执行数学计算例如 2 3 * 4。try:# 注意在生产环境中eval 有安全风险建议使用更安全的表达式解析库resulteval(expression)returnf计算结果为{result}exceptExceptionase:returnf计算错误{e}# 将工具收集到一个列表中tools[get_weather,calculate]3.2 构建 Agent 执行图LangGraphLangGraph 的核心是定义状态State、节点Node和边Edge。State一个字典用于在图的各个节点之间传递数据。Node一个函数接收 State 并返回更新后的 State。Edge定义节点之间的流转逻辑可以是条件边。# agent_graph.pyfromtypingimportTypedDict,Annotated,List,Unionfromlanggraph.graphimportStateGraph,ENDfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,AIMessage,SystemMessage,ToolMessagefromlanggraph.prebuiltimportToolExecutor,ToolInvocationimportjsonfromtoolsimporttools# 导入我们刚刚定义的工具# 1. 定义状态classAgentState(TypedDict):messages:Annotated[List[Union[HumanMessage,AIMessage,SystemMessage,ToolMessage]],对话历史]# 可以在这里添加更多自定义状态如 user_id, session_id 等# 2. 初始化 LLM 和工具执行器llmChatOpenAI(modelgpt-4o-mini,temperature0)tool_executorToolExecutor(tools)# 3. 将工具绑定到 LLM让 LLM 知道可以调用哪些工具llm_with_toolsllm.bind_tools(tools)# 4. 定义节点函数defcall_model(state:AgentState):调用 LLM 进行推理。messagesstate[messages]responsellm_with_tools.invoke(messages)return{messages:[response]}defcall_tool(state:AgentState):执行 LLM 请求的工具调用。messagesstate[messages]last_messagemessages[-1]# 解析工具调用请求tool_callslast_message.tool_calls results[]fortcintool_calls:tool_nametc[name]tool_argstc[args]# 执行工具actionToolInvocation(tooltool_name,tool_inputtool_args)responsetool_executor.invoke(action)results.append(ToolMessage(contentstr(response),tool_call_idtc[id]))return{messages:results}defshould_continue(state:AgentState):判断是否继续执行如果 LLM 返回了工具调用则继续否则结束。messagesstate[messages]last_messagemessages[-1]iflast_message.tool_calls:returncontinueelse:returnend# 5. 构建图workflowStateGraph(AgentState)# 添加节点workflow.add_node(agent,call_model)workflow.add_node(action,call_tool)# 设置入口点workflow.set_entry_point(agent)# 添加条件边workflow.add_conditional_edges(agent,should_continue,{continue:action,end:END,})# 添加从 action 回到 agent 的边形成循环workflow.add_edge(action,agent)# 编译图appworkflow.compile()4. 第三步代码实操——运行你的第一个 Agent现在让我们运行这个 Agent看看它是如何工作的。# main.pyfromagent_graphimportappfromlangchain_core.messagesimportHumanMessagedefrun_agent(user_input:str):运行 Agent 并打印结果。# 初始化状态initial_state{messages:[HumanMessage(contentuser_input)]}# 流式执行图方便观察每一步的中间结果foroutputinapp.stream(initial_state):forkey,valueinoutput.items():print(f节点 {key}:)ifmessagesinvalue:last_msgvalue[messages][-1]ifisinstance(last_msg,AIMessage)andlast_msg.content:print(f AI:{last_msg.content})elifisinstance(last_msg,ToolMessage):print(f 工具结果:{last_msg.content})print(---)# 获取最终结果final_statelist(app.stream(initial_state))[-1]final_messagefinal_state[agent][messages][-1]print(f\n最终回答:{final_message.content})if__name____main__:# 测试 1查询天气run_agent(北京和上海的天气怎么样)print(\n*50\n)# 测试 2执行计算run_agent(帮我计算 (25 17) * 3 的结果是多少)运行结果预期Agent 会先调用get_weather工具获取两个城市的天气然后整合信息回答。对于计算任务它会调用calculate工具得到结果后返回。5. 第四步调试与优化——让 Agent 更稳定生产环境中的 Agent 往往面临各种边界情况。以下是一些关键的优化策略5.1 添加错误处理与重试机制工具调用可能失败如 API 超时、返回格式错误。我们需要在call_tool节点中加入异常捕获。# 在 agent_graph.py 中优化 call_tool 函数defcall_tool(state:AgentState):messagesstate[messages]last_messagemessages[-1]tool_callslast_message.tool_calls results[]fortcintool_calls:tool_nametc[name]tool_argstc[args]try:actionToolInvocation(tooltool_name,tool_inputtool_args)responsetool_executor.invoke(action)results.append(ToolMessage(contentstr(response),tool_call_idtc[id]))exceptExceptionase:# 将错误信息返回给 LLM让它决定如何处理error_msgf工具 {tool_name} 执行失败:{str(e)}results.append(ToolMessage(contenterror_msg,tool_call_idtc[id]))return{messages:results}5.2 限制最大迭代次数防止 Agent 陷入无限循环例如反复调用工具但无法得出结论。# 在 agent_graph.py 中修改 should_continue 函数MAX_ITERATIONS10defshould_continue(state:AgentState):messagesstate[messages]# 统计已经执行了多少次工具调用tool_call_countsum(1forminmessagesifisinstance(m,ToolMessage))iftool_call_countMAX_ITERATIONS:print(f达到最大迭代次数{MAX_ITERATIONS}强制结束。)returnendlast_messagemessages[-1]iflast_message.tool_calls:returncontinueelse:returnend5.3 使用 System Prompt 约束行为通过 System Prompt 给 Agent 设定明确的角色和规则能显著提升其表现。# 在 agent_graph.py 的 call_model 函数中system_promptSystemMessage(content你是一个智能助手。你的职责是 1. 使用提供的工具来回答用户的问题。 2. 如果工具返回错误请尝试用其他方式解决或向用户说明。 3. 如果无法通过工具获取信息请诚实地告诉用户你不知道。 4. 回答要简洁、准确、友好。)defcall_model(state:AgentState):messages[system_prompt]state[messages]responsellm_with_tools.invoke(messages)return{messages:[response]}6. 第五步上线部署——将 Agent 封装为 API 服务最后我们将 Agent 封装成一个 FastAPI 服务方便前端或其他服务调用。# api.pyfromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelfromagent_graphimportappfromlangchain_core.messagesimportHumanMessage,AIMessage app_fastapiFastAPI(titleAgent Service)classQueryRequest(BaseModel):query:strsession_id:strdefault# 用于区分不同会话classQueryResponse(BaseModel):answer:strsession_id:str# 简单的会话记忆存储生产环境请使用 Redis 等session_memory{}app_fastapi.post(/agent/query,response_modelQueryResponse)asyncdefagent_query(request:QueryRequest):try:# 获取或初始化会话历史ifrequest.session_idnotinsession_memory:session_memory[request.session_id][]historysession_memory[request.session_id]# 将历史消息和当前问题一起传入input_messageshistory[HumanMessage(contentrequest.query)]initial_state{messages:input_messages}# 执行 Agentfinal_statelist(app.stream(initial_state))[-1]final_messagefinal_state[agent][messages][-1]# 更新会话历史简单起见只保留最近几轮history.append(HumanMessage(contentrequest.query))history.append(AIMessage(contentfinal_message.content))iflen(history)10:# 保留最近 5 轮对话historyhistory[-10:]session_memory[request.session_id]historyreturnQueryResponse(answerfinal_message.content,session_idrequest.session_id)exceptExceptionase:raiseHTTPException(status_code500,detailstr(e))app_fastapi.get(/health)asyncdefhealth_check():return{status:ok}# 运行方式uvicorn api:app_fastapi --reload --port 80007. 总结与最佳实践本文从零开始带你走完了 Agent 项目落地的完整流程选型选择 LangChain LangGraph 作为基础框架。架构定义了工具、状态、节点和边构建了可循环执行的 Agent 图。编码提供了可直接运行的天气查询和计算 Agent 代码。优化加入了错误处理、迭代次数限制和 System Prompt 约束。上线封装成了 FastAPI 服务并实现了简单的会话管理。落地 Agent 的几个关键建议从简单开始先让 Agent 完成一个非常具体的、边界清晰的任务再逐步扩展。工具为王Agent 的能力上限取决于你给它提供的工具。精心设计和测试每一个工具。可观测性在生产环境中务必记录 Agent 的每一步思考、工具调用和结果方便调试。人机协同对于高风险决策设计“人在回路中Human-in-the-Loop”的审批机制。希望这份指南能帮助你顺利地将 Agent 落地到你的项目中