理论讲了三篇,今天开始动手。
我们要写一个"知识库问答Agent"——用户提问,Agent检索知识库,返回答案。这是最经典的Agent场景,也是入门的最佳选择。
20分钟,从零到能运行。我们用LangGraph框架,因为它和后端工作流思维最接近。
先确认你要什么
开始之前,明确目标:
# 目标Agent的行为用户输入:"如何申请退款?"Agent:1.检测到问题意图2.调用知识库搜索工具3.检索到退款流程文档4.返回:"退款流程:订单页面 → 申请退款 → 填写原因 → 等待审核..."这是一个简单的Agent,但包含了核心要素:
- 意图识别:理解用户问什么
- 工具调用:搜索知识库
- 结果处理:返回有价值的信息
完成这个,你就掌握了Agent开发的基本套路。
第一步:环境准备(3分钟)
安装依赖:
pipinstalllanggraph langchain-openai langchain-community准备LLM。我们用OpenAI(你也可以换其他):
# config.pyimportosfromlangchain_openaiimportChatOpenAI os.environ["OPENAI_API_KEY"]="your-api-key"llm=ChatOpenAI(model="gpt-4o-mini",temperature=0)temperature=0很重要。新手Agent,我们先追求稳定性,降低随机性。
第二步:定义工具(5分钟)
Agent需要一个知识库搜索工具。我们模拟一个简单的知识库:
# tools.pyfromlangchain_core.toolsimporttool# 模拟知识库(实际项目中会连接真实数据库)KNOWLEDGE_BASE={"退款":""" 退款流程: 1. 登录账户,进入订单页面 2. 找到需要退款的订单,点击"申请退款" 3. 选择退款原因(质量问题/不想要了/其他) 4. 提交申请,等待审核(通常1-3天) 5. 审核通过后,退款将在3-5个工作日内退回原支付方式 注意:已签收超过7天的订单不支持退款 ""","发货":""" 发货说明: - 订单确认后24小时内发货 - 默认使用顺丰快递 - 发货后可在订单页面查看物流信息 - 一般时效:一线城市2-3天,其他地区3-5天 ""","会员":""" 会员权益: - VIP会员:享受10%折扣,优先发货,专属客服 - 普通会员:享受5%折扣 - 会员升级:累计消费满1000元自动升级VIP """,}@tooldefsearch_knowledge(query:str)->str:""" 搜索知识库,返回相关信息 Args: query: 搜索关键词,如"退款"、"发货"、"会员" Returns: 知识库中的相关信息,如果没有找到返回"未找到相关信息" """# 简单的关键词匹配(实际项目会用向量检索)forkeyword,contentinKNOWLEDGE_BASE.items():ifkeywordinquery:returncontent.strip()return"未找到相关信息,请尝试其他关键词或联系人工客服"@tool装饰器是LangChain的规范。每个工具必须有清晰的docstring——这是Agent理解工具用途的唯一途径。
第三步:定义Agent状态(4分钟)
LangGraph的核心是状态管理。后端开发者熟悉这个概念:
# state.pyfromtypingimportTypedDict,Annotated,Listfromlangchain_core.messagesimportBaseMessageclassAgentState(TypedDict):""" Agent的状态定义 这就像后端的Session对象,存储对话过程中的所有信息 """messages:Annotated[List[BaseMessage],"对话历史"]tool_results:Annotated[List[str],"工具调用结果"]final_answer:Annotated[str,"最终答案"]状态定义清晰,你才能追踪Agent的执行过程。这和定义API请求/响应结构一样重要。
第四步:构建Agent工作流(5分钟)
LangGraph用图来定义Agent的执行流程:
# agent.pyfromlanggraph.graphimportStateGraph,ENDfromlangchain_core.messagesimportHumanMessage,AIMessagefromlangchain_core.promptsimportChatPromptTemplatefromconfigimportllmfromtoolsimportsearch_knowledgefromstateimportAgentState# Prompt定义prompt=ChatPromptTemplate.from_messages([("system","""你是客服助手Agent。 当用户提问时,先判断问题类型,然后使用search_knowledge工具搜索相关信息。 如果找到相关信息,整理后返回给用户。 如果没有找到,告知用户并建议联系人工客服。 回答时要简洁清晰,不要冗长。 """),("human","{input}"),])# 节点:调用LLM决策defagent_node(state:AgentState)->AgentState:"""Agent决策节点:分析用户输入,决定下一步"""user_input=state["messages"][-1].contentifstate["messages"]else""# 让LLM绑定工具,进行决策llm_with_tools=llm.bind_tools([search_knowledge])response=llm_with_tools.invoke(prompt.format(input=user_input))# 更新状态state["messages"].append(response)returnstate# 节点:执行工具调用deftool_node(state:AgentState)->AgentState:"""工具执行节点:调用search_knowledge工具"""last_message=state["messages"][-1]# 检查是否有工具调用请求ifhasattr(last_message,"tool_calls")andlast_message.tool_calls:fortool_callinlast_message.tool_calls:iftool_call["name"]=="search_knowledge":result=search_knowledge.invoke(tool_call["args"])state["tool_results"].append(result)# 把工具结果作为消息加入对话历史state["messages"].append(AIMessage(content=f"工具返回:{result}"))returnstate# 节点:生成最终答案defanswer_node(state:AgentState)->AgentState:"""答案生成节点:根据工具结果,生成最终回复"""tool_results=state["tool_results"]user_question=state["messages"][0].contentifstate["messages"]else""iftool_results:# 有工具结果,整理后回答context="\n".join(tool_results)answer_prompt=f"""根据以下信息回答用户问题: 信息:{context}用户问题:{user_question}请简洁清晰地回答。"""response=llm.invoke(answer_prompt)state["final_answer"]=response.contentelse:# 没有工具调用,直接用LLM回答state["final_answer"]=state["messages"][-1].contentreturnstate# 路由函数:决定下一步defshould_call_tool(state:AgentState)->str:"""判断是否需要调用工具"""last_message=state["messages"][-1]ifhasattr(last_message,"tool_calls")andlast_message.tool_calls:return"tool"else:return"answer"# 构建工作流图workflow=StateGraph(AgentState)# 添加节点workflow.add_node("agent",agent_node)workflow.add_node("tool",tool_node)workflow.add_node("answer",answer_node)# 设置入口workflow.set_entry_point("agent")# 添加边(工作流路径)workflow.add_conditional_edges("agent",should_call_tool,{"tool":"tool","answer":"answer"})workflow.add_edge("tool","agent")# 工具执行后回到Agent决策workflow.add_edge("answer",END)# 答案生成后结束# 编译工作流app=workflow.compile()这个工作流图和后端开发的流程图很像:
入口 → Agent决策 → 是否调用工具? ↓是 ↓否 工具执行 → Agent决策 → 答案生成 → 结束每一步都是明确的节点,状态在各节点间传递。
第五步:运行Agent(3分钟)
# main.pyfromlangchain_core.messagesimportHumanMessagefromagentimportappfromstateimportAgentStatedefrun_agent(user_input:str)->str:"""运行Agent,返回答案"""initial_state=AgentState(messages=[HumanMessage(content=user_input)],tool_results=[],final_answer="")# 执行工作流result=app.invoke(initial_state)returnresult["final_answer"]# 测试运行if__name__=="__main__":questions=["怎么退款?","发货需要多久?","VIP会员有什么权益?","我想投诉",]forquestioninquestions:print(f"\n用户:{question}")answer=run_agent(question)print(f"Agent:{answer}")运行效果:
用户:怎么退款? Agent:退款流程如下: 1. 登录账户,进入订单页面 2. 找到需要退款的订单,点击"申请退款" 3. 选择退款原因 4. 提交申请,等待审核(通常1-3天) 5. 审核通过后,退款将在3-5个工作日内退回原支付方式 注意:已签收超过7天的订单不支持退款 用户:发货需要多久? Agent:发货说明: - 订单确认后24小时内发货 - 一线城市2-3天送达,其他地区3-5天 - 可在订单页面查看物流信息 用户:VIP会员有什么权益? Agent:VIP会员权益: - 享受10%折扣 - 优先发货 - 专属客服通道 - 累计消费满1000元可自动升级VIP 用户:我想投诉 Agent:未找到相关信息,建议您联系人工客服处理投诉事宜。你的第一个Agent已经能用了。
看看Agent做了什么
我们加个追踪,看看Agent内部的决策过程:
# debug.pyfromagentimportappfromstateimportAgentStatefromlangchain_core.messagesimportHumanMessagedefdebug_agent(user_input:str):"""调试Agent,查看内部过程"""state=AgentState(messages=[HumanMessage(content=user_input)],tool_results=[],final_answer="")# LangGraph支持step-by-step追踪forstepinapp.stream(state):node_name=list(step.keys())[0]node_state=step[node_name]print(f"\n==={node_name}节点 ===")ifnode_name=="agent":last_msg=node_state["messages"][-1]print(f"Agent思考:{last_msg.content[:100]}...")ifhasattr(last_msg,"tool_calls")andlast_msg.tool_calls:print(f"决定调用工具:{last_msg.tool_calls}")elifnode_name=="tool":print(f"工具结果:{node_state['tool_results']}")elifnode_name=="answer":print(f"最终答案:{node_state['final_answer']}")debug_agent("怎么退款?")输出:
=== agent节点 === Agent思考:... 决定调用工具:[{'name': 'search_knowledge', 'args': {'query': '退款'}}] === tool节点 === 工具结果:['退款流程:...'] === agent节点 === Agent思考:根据工具返回的信息,用户想知道退款流程... (没有工具调用,准备回答) === answer节点 === 最终答案:退款流程如下...你能看到Agent的完整决策链路:
- 第一次agent节点:检测到退款问题,决定调用search_knowledge
- tool节点:执行工具,返回知识库内容
- 第二次agent节点:拿到工具结果,决定直接回答(不再调用工具)
- answer节点:整理信息,生成最终答案
这个追踪能力,和后端日志追踪一样重要。
几个设计要点
完成第一个Agent后,回顾几个关键设计:
1. 工具定义要有明确边界
@tooldefsearch_knowledge(query:str)->str:""" 搜索知识库,返回相关信息 Returns: 知识库中的相关信息,如果没有找到返回"未找到相关信息" """返回格式明确:要么返回信息,要么返回"未找到相关信息"。不要返回空字符串或None——LLM无法理解。
2. Prompt要有执行约束
prompt=""" 你是客服助手Agent。 当用户提问时,先判断问题类型,然后使用search_knowledge工具搜索相关信息。 如果找到相关信息,整理后返回给用户。 如果没有找到,告知用户并建议联系人工客服。 """“先判断、再搜索、再回答”——这个顺序约束防止Agent跳过关键步骤。
3. 状态要有生命周期
classAgentState(TypedDict):messages:List[BaseMessage]# 对话历史tool_results:List[str]# 工具结果(本轮)final_answer:str# 最终答案tool_results是本轮的工具结果,不是累积的历史。每轮对话开始时清空,防止状态污染。
4. 工作流要有终止条件
workflow.add_edge("answer",END)# 明确的终止Agent不会无限循环。answer节点执行后,直接END。
你可以继续扩展
这个Agent是基础版本,你可以继续扩展:
扩展1:添加更多工具
@tooldefcheck_order(order_id:str)->str:"""查询订单状态"""...@tooldeftransfer_to_human(reason:str)->str:"""转人工客服"""...# Agent的tools列表更新llm_with_tools=llm.bind_tools([search_knowledge,check_order,transfer_to_human])扩展2:支持多轮对话
defrun_multi_turn_conversation():"""多轮对话模式"""state=AgentState(messages=[],tool_results=[],final_answer="")whileTrue:user_input=input("用户:")ifuser_input=="exit":breakstate["messages"].append(HumanMessage(content=user_input))state["tool_results"]=[]# 清空本轮工具结果result=app.invoke(state)state=result# 更新状态,保留对话历史print(f"Agent:{state['final_answer']}")扩展3:使用真实知识库
# 用向量数据库替代模拟数据fromlangchain_community.vectorstoresimportChromafromlangchain_openaiimportOpenAIEmbeddings vectorstore=Chroma.from_documents(documents=your_documents,embedding=OpenAIEmbeddings())@tooldefsearch_knowledge(query:str)->str:"""向量检索知识库"""results=vectorstore.similarity_search(query,k=3)return"\n".join([doc.page_contentfordocinresults])回顾这20分钟
你完成了:
- 环境准备:安装LangGraph,配置LLM
- 工具定义:search_knowledge工具,清晰的输入输出
- 状态定义:AgentState,管理对话信息
- 工作流构建:agent→tool→answer→END的执行路径
- 运行验证:Agent能正确回答问题
更重要的是,你用后端思维理解了Agent:
- 工具 = 外部服务调用
- 状态 = Session管理
- 工作流 = 业务流程编排
- Prompt = 接口定义(自然语言版)
下一篇文章,我们聊框架选型:LangChain vs LangGraph vs CrewAI,帮你理解不同框架的设计理念和适用场景。