企业官网建设流程全解析

1. 引言

大语言模型（LLM）的能力已经毋庸置疑，但如何让它在真实的业务场景中稳定、可靠地完成任务，而不是仅仅做一个“聊天机器人”？答案就是Agent（智能体）。

Agent 的核心在于让 LLM 具备感知、规划、行动和记忆的能力，从而自主地完成复杂任务。然而，从概念到生产落地，中间隔着“选型、架构、开发、调试、上线”五座大山。

本文将带你手把手走完这一完整流程，从技术选型开始，到最终部署一个能稳定运行的 Agent 服务，并提供可直接运行的代码示例。

2. 第一步：技术选型——选择合适的 Agent 框架

在开始编码前，我们需要选择一个合适的框架。目前主流的 Agent 框架有 LangChain、AutoGPT、CrewAI 等。对于企业级项目落地，LangChain和LangGraph是目前最成熟、生态最完善的选择。

2.1 为什么选择 LangChain + LangGraph？

• LangChain：提供了丰富的 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 架构包含以下几个核心组件：

1. LLM（大脑）：负责推理、规划和决策。
2. Tools（手脚）：Agent 可以调用的外部能力，如搜索引擎、数据库查询、API 调用、代码执行器等。
3. Memory（记忆）：记录对话历史和中间状态，让 Agent 具备上下文理解能力。
4. Planning（规划）：将复杂任务分解为可执行的步骤。

3.1 定义工具（Tools）

工具是 Agent 与外部世界交互的桥梁。我们以两个常见场景为例：获取天气和执行数学计算。

# tools.pyfromlangchain_core.toolsimporttoolimportrequests@tooldefget_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 有安全风险，建议使用更安全的表达式解析库result=eval(expression)returnf"计算结果为：{result}"exceptExceptionase:returnf"计算错误：{e}"# 将工具收集到一个列表中tools=[get_weather,calculate]

3.2 构建 Agent 执行图（LangGraph）

LangGraph 的核心是定义状态（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 和工具执行器llm=ChatOpenAI(model="gpt-4o-mini",temperature=0)tool_executor=ToolExecutor(tools)# 3. 将工具绑定到 LLM，让 LLM 知道可以调用哪些工具llm_with_tools=llm.bind_tools(tools)# 4. 定义节点函数defcall_model(state:AgentState):"""调用 LLM 进行推理。"""messages=state["messages"]response=llm_with_tools.invoke(messages)return{"messages":[response]}defcall_tool(state:AgentState):"""执行 LLM 请求的工具调用。"""messages=state["messages"]last_message=messages[-1]# 解析工具调用请求tool_calls=last_message.tool_calls results=[]fortcintool_calls:tool_name=tc["name"]tool_args=tc["args"]# 执行工具action=ToolInvocation(tool=tool_name,tool_input=tool_args)response=tool_executor.invoke(action)results.append(ToolMessage(content=str(response),tool_call_id=tc["id"]))return{"messages":results}defshould_continue(state:AgentState):"""判断是否继续执行（如果 LLM 返回了工具调用，则继续；否则结束）。"""messages=state["messages"]last_message=messages[-1]iflast_message.tool_calls:return"continue"else:return"end"# 5. 构建图workflow=StateGraph(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")# 编译图app=workflow.compile()

4. 第三步：代码实操——运行你的第一个 Agent

现在，让我们运行这个 Agent，看看它是如何工作的。

# main.pyfromagent_graphimportappfromlangchain_core.messagesimportHumanMessagedefrun_agent(user_input:str):"""运行 Agent 并打印结果。"""# 初始化状态initial_state={"messages":[HumanMessage(content=user_input)]}# 流式执行图，方便观察每一步的中间结果foroutputinapp.stream(initial_state):forkey,valueinoutput.items():print(f"节点 '{key}':")if"messages"invalue:last_msg=value["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_state=list(app.stream(initial_state))[-1]final_message=final_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):messages=state["messages"]last_message=messages[-1]tool_calls=last_message.tool_calls results=[]fortcintool_calls:tool_name=tc["name"]tool_args=tc["args"]try:action=ToolInvocation(tool=tool_name,tool_input=tool_args)response=tool_executor.invoke(action)results.append(ToolMessage(content=str(response),tool_call_id=tc["id"]))exceptExceptionase:# 将错误信息返回给 LLM，让它决定如何处理error_msg=f"工具 '{tool_name}' 执行失败:{str(e)}"results.append(ToolMessage(content=error_msg,tool_call_id=tc["id"]))return{"messages":results}

5.2 限制最大迭代次数

防止 Agent 陷入无限循环（例如，反复调用工具但无法得出结论）。

# 在 agent_graph.py 中修改 should_continue 函数MAX_ITERATIONS=10defshould_continue(state:AgentState):messages=state["messages"]# 统计已经执行了多少次工具调用tool_call_count=sum(1forminmessagesifisinstance(m,ToolMessage))iftool_call_count>=MAX_ITERATIONS:print(f"达到最大迭代次数{MAX_ITERATIONS}，强制结束。")return"end"last_message=messages[-1]iflast_message.tool_calls:return"continue"else:return"end"

5.3 使用 System Prompt 约束行为

通过 System Prompt 给 Agent 设定明确的角色和规则，能显著提升其表现。

# 在 agent_graph.py 的 call_model 函数中system_prompt=SystemMessage(content="""你是一个智能助手。你的职责是： 1. 使用提供的工具来回答用户的问题。 2. 如果工具返回错误，请尝试用其他方式解决或向用户说明。 3. 如果无法通过工具获取信息，请诚实地告诉用户你不知道。 4. 回答要简洁、准确、友好。""")defcall_model(state:AgentState):messages=[system_prompt]+state["messages"]response=llm_with_tools.invoke(messages)return{"messages":[response]}

6. 第五步：上线部署——将 Agent 封装为 API 服务

最后，我们将 Agent 封装成一个 FastAPI 服务，方便前端或其他服务调用。

# api.pyfromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelfromagent_graphimportappfromlangchain_core.messagesimportHumanMessage,AIMessage app_fastapi=FastAPI(title="Agent Service")classQueryRequest(BaseModel):query:strsession_id:str="default"# 用于区分不同会话classQueryResponse(BaseModel):answer:strsession_id:str# 简单的会话记忆存储（生产环境请使用 Redis 等）session_memory={}@app_fastapi.post("/agent/query",response_model=QueryResponse)asyncdefagent_query(request:QueryRequest):try:# 获取或初始化会话历史ifrequest.session_idnotinsession_memory:session_memory[request.session_id]=[]history=session_memory[request.session_id]# 将历史消息和当前问题一起传入input_messages=history+[HumanMessage(content=request.query)]initial_state={"messages":input_messages}# 执行 Agentfinal_state=list(app.stream(initial_state))[-1]final_message=final_state["agent"]["messages"][-1]# 更新会话历史（简单起见，只保留最近几轮）history.append(HumanMessage(content=request.query))history.append(AIMessage(content=final_message.content))iflen(history)>10:# 保留最近 5 轮对话history=history[-10:]session_memory[request.session_id]=historyreturnQueryResponse(answer=final_message.content,session_id=request.session_id)exceptExceptionase:raiseHTTPException(status_code=500,detail=str(e))@app_fastapi.get("/health")asyncdefhealth_check():return{"status":"ok"}# 运行方式：uvicorn api:app_fastapi --reload --port 8000

7. 总结与最佳实践

本文从零开始，带你走完了 Agent 项目落地的完整流程：

1. 选型：选择 LangChain + LangGraph 作为基础框架。
2. 架构：定义了工具、状态、节点和边，构建了可循环执行的 Agent 图。
3. 编码：提供了可直接运行的天气查询和计算 Agent 代码。
4. 优化：加入了错误处理、迭代次数限制和 System Prompt 约束。
5. 上线：封装成了 FastAPI 服务，并实现了简单的会话管理。

落地 Agent 的几个关键建议：

• 从简单开始：先让 Agent 完成一个非常具体的、边界清晰的任务，再逐步扩展。
• 工具为王：Agent 的能力上限取决于你给它提供的工具。精心设计和测试每一个工具。
• 可观测性：在生产环境中，务必记录 Agent 的每一步思考、工具调用和结果，方便调试。
• 人机协同：对于高风险决策，设计“人在回路中（Human-in-the-Loop）”的审批机制。

希望这份指南能帮助你顺利地将 Agent 落地到你的项目中！