第十六天后端代码学习笔记1 / 2 Agent 入门案例 ReAct 模式先看case14.py这是最基础的 agent 长啥样。importosfromdatetimeimportdatetimefromlangchain.agentsimportcreate_agentfromlangchain_core.toolsimporttoolfromlangchain_openaiimportChatOpenAI# 1:定义大模型llmChatOpenAI(modelqwen-plus,temperature0.75,api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://ws-89t66lrlxx0rdobw.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)先把大模型接上。这里用的是 LangChain 的ChatOpenAI参数跟咱们之前直接调 OpenAI SDK 差不多base_url还是那个阿里云百炼的地址model用qwen-plus。然后定义工具这块跟之前手搓的不一样了用的是装饰器# 2:定义工具,一定放到工具列表中tool(description获取指定城市的当前时间,当用户询问时间、现在几点、当前时间等问题时调用此工具。 参数: city,字符串类型,城市,比如:北京,上海 返回值:当前城市的时间字符串)defget_current_time(city:str)-str:now_strdatetime.now().strftime(%Y年%m月%d日 %H%M%S)returnf{city}的当前时间是{now_str}tool这个装饰器挺省事的函数名就是工具名参数类型标注就是入参函数文档字符串docstring或者description就是给大模型看的工具说明。大模型就靠这个description来判断用户问的是时间那我得调get_current_time。下面那个get_current_weather同理就是个写死的假数据字典查不到就返回提示。关键是这一段——把模型、工具、人设拼成一个 agent# 3:创建agent- ReAct[思考-行动-观察]system_prompt 你是一个智能助手 你可以使用工具来解决问题 工具列表: 1:当用户询问时间、现在几点、当前时间等问题时,调用get_current_time工具 2:当用户询问天气,气温等问题时,调用get_current_weather工具 请根据用户的问题,自主决定是否调用工具,调用哪个工具,调用顺序 如果不需要工具,直接回答用户的问题 agentcreate_agent(modelllm,toolstools,system_promptsystem_prompt,debugTrue,)create_agent一句话就把活干完了。你要是问之前自己手写的那个 function calling 流程定义 tools、判断 tool_calls、塞回 messages、再调一次这里全给你包里面了。ReAct是啥其实就是 agent 干活的逻辑思考Reasoning→ 行动Action→ 观察Observation循环。用户问北京天气它先思考这得查天气工具然后行动调get_current_weather(北京)拿到结果后观察再组织语言回你。开了debugTrue你能看见它每一步在想啥挺好玩的。最后跑起来# 4:执行agentresultagent.invoke({messages:[{role:user,content:北京的天气是什么}]})print(result)print(result.get(messages)[-1].content)agent.invoke一调里面自己蹦跶好几轮最后从消息列表里取最后一条 AI 消息的content就是答案。3 Checkpointer 机制短期记忆case14.py有个毛病它记不住你上一轮说过啥。比如你先说我叫张三喜欢北京再问我最喜欢的城市天气咋样它就懵了因为每回invoke都是全新的没有上下文。case15.py就来解决这个加了个记忆。fromlanggraph.checkpoint.memoryimportInMemorySaver# 实例化记忆存储器这里用内存存储,也支持AsyncPostgresSavercheckpointerInMemorySaver()agentcreate_agent(modelllm,toolstools,system_promptsystem_prompt,checkpointercheckpointer,# 检查点,海马体,用于保存当前的状态,短期记忆debugTrue,)InMemorySaver就是个内存版的记忆存储注释里管它叫海马体挺形象的。它是短期记忆——进程不关就一直记着关了就没了要持久化得换AsyncPostgresSaver那种接数据库的。光有 checkpointer 不够还得告诉它这是谁的记忆if__name____main__:print(\n [第1轮] 用户: 我叫user_001我最喜欢的城市是上海我喜欢晴天。)thread_iduser_001config{configurable:{thread_id:thread_id,}}res1agent.invoke(input{messages:[{role:user,content:我叫user_001我最喜欢的城市是上海我喜欢晴天。}]},configconfig)这个thread_id就是会话的身份标识相当于给每个用户发一个笔记本。下次同一个thread_id再来问agent 就从 checkpointer 里把之前那本笔记翻出来知道你叫 user_001、喜欢上海。不同thread_id之间互不干扰各记各的。这其实就是咱们之前用 Redis 存多轮对话那套思路的官方版封装LangChain 帮你管好了。4 结构化输出有时候你不想让模型返回一大段自然语言文字而是想要个固定格式比如姓名、邮箱、手机号拆开的好填进数据库。case16.py就干这个。importosfromlangchain_openaiimportChatOpenAIfrompydanticimportBaseModel,Fieldfromlangchain.agentsimportcreate_agent llmChatOpenAI(modelqwen-plus,temperature0.75,api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://ws-89t66lrlxx0rdobw.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)classContactInfo(BaseModel):name:strField(description姓名)email:strField(description个人邮箱)phone:strField(description个人手机号)agentcreate_agent(modelllm,response_formatContactInfo)这里用pydantic的BaseModel定义了一个联系人信息的结构Field里写的是字段说明。然后create_agent里传response_formatContactInfo等于告诉模型你就按这个模板给我填别瞎写。跑的时候resultagent.invoke({messages:[{role:user,content:我的联系方式:张三,zhangsanqq.com,13800138000}]})print(result[structured_response])出来的structured_response就是个规整的对象.name、.email、.phone直接能取不用再去正则抠一段文本里的手机号了。这玩意儿对接咱们 boss_api 那种要落库的场景特别香。5 / 6 中间件Middleware实际跑 agent经常会出幺蛾子模型偶尔抽风返回格式不对、工具调用失败、或者用户一直聊一直聊把上下文撑爆。case17.py就把这些保命逻辑用中间件串上。fromlangchain.agents.middlewareimportModelRetryMiddleware,ToolRetryMiddleware,ModelCallLimitMiddleware,\ SummarizationMiddlewarefromlanggraph.checkpoint.memoryimportInMemorySaver# ... 前面定义 llm、tools、system_prompt 跟 case15 一样 ...checkpointerInMemorySaver()agentcreate_agent(modelllm,toolstools,system_promptsystem_prompt,checkpointercheckpointer,debugTrue,middleware[ModelRetryMiddleware(max_retries3),ToolRetryMiddleware(max_retries3),ModelCallLimitMiddleware(run_limit10),SummarizationMiddleware()])挨个说这几个中间件是干嘛的ModelRetryMiddleware(max_retries3)模型调用失败自动重试最多 3 次。比如网络抖了一下返回 500它自己重来不用人管。ToolRetryMiddleware(max_retries3)工具执行失败也重试比如调天气接口超时了重调。ModelCallLimitMiddleware(run_limit10)限制一轮对话里模型最多被调 10 次。防止 agent 犯轴在一个死循环里反复调模型把 token 烧光也防它一直转圈不出来。SummarizationMiddleware()聊太长了自动把前面的历史压缩成摘要腾出上下文空间。就是咱们之前在 case2_api 里手写的那个滑动窗口历史压缩的官方版。这些中间件一层套一层agent 跑的时候自动生效。case17.py底部也留了个thread_id的例子跟 17-3 一样的用法只是这次带着中间件一起跑更稳。7 / 8 LangChain 的缺点 LangGraph 入门概念补充讲真到 17-7、17-8 老师开始讲 LangGraph 了但我扒了一圈boss_api这个项目目前llm/目录下只有case14~17这几个文件没看到专门写StateGraph的入门案例源码grep 只在 case15、case17 里命中了langgraph的 checkpoint 导入那是 create_agent 底层依赖的不是咱们手写的图。所以这块我先按老师课上讲的意思给你捋一下等源码放进来再补。为啥要 LangGraphcreate_agent好用是真好用但它把流程封装死了——就是个标准的 ReAct 循环你想改流程比如先查数据库再让两个 agent 分工最后汇总就很不灵活。langchain早期那种链Chain又太死板。LangGraph 的思路是把 agent 跑的流程画成一张图Graph节点Node是每一步干啥边Edge是走哪条路还能根据条件分支。这样复杂流程你自己画想怎么连怎么连。LangGraph 入门大概长这样通用写法项目里暂时没这文件fromlanggraph.graphimportStateGraph,MessagesState,START,ENDfromlanggraph.prebuiltimportToolNode# 1. 定义图状态用 MessagesState就是消息列表builderStateGraph(MessagesState)# 2. 加节点一个节点是调模型一个节点是执行工具builder.add_node(model,call_model)builder.add_node(tools,ToolNode(tools))# 3. 加边START - modeltools - model工具跑完回到模型builder.add_edge(START,model)builder.add_edge(tools,model)# 4. 条件边model 之后看有没有 tool_calls有就走 tools没有就 ENDbuilder.add_conditional_edges(model,should_continue,{tools:tools,END:END})# 5. 编译出可执行的图graphbuilder.compile()核心就三样东西节点add_node、普通边add_edge按顺序走、条件边add_conditional_edges根据模型输出决定下一步去哪。把这套学会比create_agent那种黑盒灵活太多。等老师把 17-8 的 LangGraph 入门案例代码发下来或者你自己跟着敲一个case18.py我再给你按同样的口味讲一遍。 第十六天后端代码学习笔记
第十六天后端代码学习笔记1 / 2 Agent 入门案例 ReAct 模式先看case14.py这是最基础的 agent 长啥样。importosfromdatetimeimportdatetimefromlangchain.agentsimportcreate_agentfromlangchain_core.toolsimporttoolfromlangchain_openaiimportChatOpenAI# 1:定义大模型llmChatOpenAI(modelqwen-plus,temperature0.75,api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://ws-89t66lrlxx0rdobw.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)先把大模型接上。这里用的是 LangChain 的ChatOpenAI参数跟咱们之前直接调 OpenAI SDK 差不多base_url还是那个阿里云百炼的地址model用qwen-plus。然后定义工具这块跟之前手搓的不一样了用的是装饰器# 2:定义工具,一定放到工具列表中tool(description获取指定城市的当前时间,当用户询问时间、现在几点、当前时间等问题时调用此工具。 参数: city,字符串类型,城市,比如:北京,上海 返回值:当前城市的时间字符串)defget_current_time(city:str)-str:now_strdatetime.now().strftime(%Y年%m月%d日 %H%M%S)returnf{city}的当前时间是{now_str}tool这个装饰器挺省事的函数名就是工具名参数类型标注就是入参函数文档字符串docstring或者description就是给大模型看的工具说明。大模型就靠这个description来判断用户问的是时间那我得调get_current_time。下面那个get_current_weather同理就是个写死的假数据字典查不到就返回提示。关键是这一段——把模型、工具、人设拼成一个 agent# 3:创建agent- ReAct[思考-行动-观察]system_prompt 你是一个智能助手 你可以使用工具来解决问题 工具列表: 1:当用户询问时间、现在几点、当前时间等问题时,调用get_current_time工具 2:当用户询问天气,气温等问题时,调用get_current_weather工具 请根据用户的问题,自主决定是否调用工具,调用哪个工具,调用顺序 如果不需要工具,直接回答用户的问题 agentcreate_agent(modelllm,toolstools,system_promptsystem_prompt,debugTrue,)create_agent一句话就把活干完了。你要是问之前自己手写的那个 function calling 流程定义 tools、判断 tool_calls、塞回 messages、再调一次这里全给你包里面了。ReAct是啥其实就是 agent 干活的逻辑思考Reasoning→ 行动Action→ 观察Observation循环。用户问北京天气它先思考这得查天气工具然后行动调get_current_weather(北京)拿到结果后观察再组织语言回你。开了debugTrue你能看见它每一步在想啥挺好玩的。最后跑起来# 4:执行agentresultagent.invoke({messages:[{role:user,content:北京的天气是什么}]})print(result)print(result.get(messages)[-1].content)agent.invoke一调里面自己蹦跶好几轮最后从消息列表里取最后一条 AI 消息的content就是答案。3 Checkpointer 机制短期记忆case14.py有个毛病它记不住你上一轮说过啥。比如你先说我叫张三喜欢北京再问我最喜欢的城市天气咋样它就懵了因为每回invoke都是全新的没有上下文。case15.py就来解决这个加了个记忆。fromlanggraph.checkpoint.memoryimportInMemorySaver# 实例化记忆存储器这里用内存存储,也支持AsyncPostgresSavercheckpointerInMemorySaver()agentcreate_agent(modelllm,toolstools,system_promptsystem_prompt,checkpointercheckpointer,# 检查点,海马体,用于保存当前的状态,短期记忆debugTrue,)InMemorySaver就是个内存版的记忆存储注释里管它叫海马体挺形象的。它是短期记忆——进程不关就一直记着关了就没了要持久化得换AsyncPostgresSaver那种接数据库的。光有 checkpointer 不够还得告诉它这是谁的记忆if__name____main__:print(\n [第1轮] 用户: 我叫user_001我最喜欢的城市是上海我喜欢晴天。)thread_iduser_001config{configurable:{thread_id:thread_id,}}res1agent.invoke(input{messages:[{role:user,content:我叫user_001我最喜欢的城市是上海我喜欢晴天。}]},configconfig)这个thread_id就是会话的身份标识相当于给每个用户发一个笔记本。下次同一个thread_id再来问agent 就从 checkpointer 里把之前那本笔记翻出来知道你叫 user_001、喜欢上海。不同thread_id之间互不干扰各记各的。这其实就是咱们之前用 Redis 存多轮对话那套思路的官方版封装LangChain 帮你管好了。4 结构化输出有时候你不想让模型返回一大段自然语言文字而是想要个固定格式比如姓名、邮箱、手机号拆开的好填进数据库。case16.py就干这个。importosfromlangchain_openaiimportChatOpenAIfrompydanticimportBaseModel,Fieldfromlangchain.agentsimportcreate_agent llmChatOpenAI(modelqwen-plus,temperature0.75,api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://ws-89t66lrlxx0rdobw.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)classContactInfo(BaseModel):name:strField(description姓名)email:strField(description个人邮箱)phone:strField(description个人手机号)agentcreate_agent(modelllm,response_formatContactInfo)这里用pydantic的BaseModel定义了一个联系人信息的结构Field里写的是字段说明。然后create_agent里传response_formatContactInfo等于告诉模型你就按这个模板给我填别瞎写。跑的时候resultagent.invoke({messages:[{role:user,content:我的联系方式:张三,zhangsanqq.com,13800138000}]})print(result[structured_response])出来的structured_response就是个规整的对象.name、.email、.phone直接能取不用再去正则抠一段文本里的手机号了。这玩意儿对接咱们 boss_api 那种要落库的场景特别香。5 / 6 中间件Middleware实际跑 agent经常会出幺蛾子模型偶尔抽风返回格式不对、工具调用失败、或者用户一直聊一直聊把上下文撑爆。case17.py就把这些保命逻辑用中间件串上。fromlangchain.agents.middlewareimportModelRetryMiddleware,ToolRetryMiddleware,ModelCallLimitMiddleware,\ SummarizationMiddlewarefromlanggraph.checkpoint.memoryimportInMemorySaver# ... 前面定义 llm、tools、system_prompt 跟 case15 一样 ...checkpointerInMemorySaver()agentcreate_agent(modelllm,toolstools,system_promptsystem_prompt,checkpointercheckpointer,debugTrue,middleware[ModelRetryMiddleware(max_retries3),ToolRetryMiddleware(max_retries3),ModelCallLimitMiddleware(run_limit10),SummarizationMiddleware()])挨个说这几个中间件是干嘛的ModelRetryMiddleware(max_retries3)模型调用失败自动重试最多 3 次。比如网络抖了一下返回 500它自己重来不用人管。ToolRetryMiddleware(max_retries3)工具执行失败也重试比如调天气接口超时了重调。ModelCallLimitMiddleware(run_limit10)限制一轮对话里模型最多被调 10 次。防止 agent 犯轴在一个死循环里反复调模型把 token 烧光也防它一直转圈不出来。SummarizationMiddleware()聊太长了自动把前面的历史压缩成摘要腾出上下文空间。就是咱们之前在 case2_api 里手写的那个滑动窗口历史压缩的官方版。这些中间件一层套一层agent 跑的时候自动生效。case17.py底部也留了个thread_id的例子跟 17-3 一样的用法只是这次带着中间件一起跑更稳。7 / 8 LangChain 的缺点 LangGraph 入门概念补充讲真到 17-7、17-8 老师开始讲 LangGraph 了但我扒了一圈boss_api这个项目目前llm/目录下只有case14~17这几个文件没看到专门写StateGraph的入门案例源码grep 只在 case15、case17 里命中了langgraph的 checkpoint 导入那是 create_agent 底层依赖的不是咱们手写的图。所以这块我先按老师课上讲的意思给你捋一下等源码放进来再补。为啥要 LangGraphcreate_agent好用是真好用但它把流程封装死了——就是个标准的 ReAct 循环你想改流程比如先查数据库再让两个 agent 分工最后汇总就很不灵活。langchain早期那种链Chain又太死板。LangGraph 的思路是把 agent 跑的流程画成一张图Graph节点Node是每一步干啥边Edge是走哪条路还能根据条件分支。这样复杂流程你自己画想怎么连怎么连。LangGraph 入门大概长这样通用写法项目里暂时没这文件fromlanggraph.graphimportStateGraph,MessagesState,START,ENDfromlanggraph.prebuiltimportToolNode# 1. 定义图状态用 MessagesState就是消息列表builderStateGraph(MessagesState)# 2. 加节点一个节点是调模型一个节点是执行工具builder.add_node(model,call_model)builder.add_node(tools,ToolNode(tools))# 3. 加边START - modeltools - model工具跑完回到模型builder.add_edge(START,model)builder.add_edge(tools,model)# 4. 条件边model 之后看有没有 tool_calls有就走 tools没有就 ENDbuilder.add_conditional_edges(model,should_continue,{tools:tools,END:END})# 5. 编译出可执行的图graphbuilder.compile()核心就三样东西节点add_node、普通边add_edge按顺序走、条件边add_conditional_edges根据模型输出决定下一步去哪。把这套学会比create_agent那种黑盒灵活太多。等老师把 17-8 的 LangGraph 入门案例代码发下来或者你自己跟着敲一个case18.py我再给你按同样的口味讲一遍。