企业官网建设流程全解析

第十六天后端代码学习笔记

1 / 2 Agent 入门案例 + ReAct 模式

先看case14.py，这是最基础的 agent 长啥样。

importosfromdatetimeimportdatetimefromlangchain.agentsimportcreate_agentfromlangchain_core.toolsimporttoolfromlangchain_openaiimportChatOpenAI# 1:定义大模型llm=ChatOpenAI(model='qwen-plus',temperature=0.75,api_key=os.getenv("DASHSCOPE_API_KEY"),base_url="https://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_str=datetime.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工具 请根据用户的问题,自主决定是否调用工具,调用哪个工具,调用顺序 如果不需要工具,直接回答用户的问题 """agent=create_agent(model=llm,tools=tools,system_prompt=system_prompt,debug=True,)

create_agent一句话就把活干完了。你要是问之前自己手写的那个 function calling 流程（定义 tools、判断 tool_calls、塞回 messages、再调一次），这里全给你包里面了。

ReAct是啥？其实就是 agent 干活的逻辑：思考（Reasoning）→ 行动（Action）→ 观察（Observation），循环。用户问"北京天气"，它先思考"这得查天气工具"，然后行动"调get_current_weather('北京')"，拿到结果后观察，再组织语言回你。开了debug=True你能看见它每一步在想啥，挺好玩的。

最后跑起来：

# 4:执行agentresult=agent.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# 实例化记忆存储器（这里用内存存储）,也支持AsyncPostgresSavercheckpointer=InMemorySaver()agent=create_agent(model=llm,tools=tools,system_prompt=system_prompt,checkpointer=checkpointer,# 检查点,海马体,用于保存当前的状态,短期记忆debug=True,)

InMemorySaver就是个内存版的记忆存储，注释里管它叫"海马体"，挺形象的。它是短期记忆——进程不关就一直记着，关了就没了（要持久化得换AsyncPostgresSaver那种接数据库的）。

光有 checkpointer 不够，还得告诉它"这是谁的记忆"：

if__name__=='__main__':print("\n📝 [第1轮] 用户: 我叫user_001，我最喜欢的城市是上海，我喜欢晴天。")thread_id="user_001"config={"configurable":{"thread_id":thread_id,}}res1=agent.invoke(input={"messages":[{"role":"user","content":"我叫user_001，我最喜欢的城市是上海，我喜欢晴天。"}]},config=config)

这个thread_id就是会话的身份标识，相当于给每个用户发一个笔记本。下次同一个thread_id再来问，agent 就从 checkpointer 里把之前那本笔记翻出来，知道你叫 user_001、喜欢上海。不同thread_id之间互不干扰，各记各的。

这其实就是咱们之前用 Redis 存多轮对话那套思路的"官方版"封装，LangChain 帮你管好了。

4 结构化输出

有时候你不想让模型返回一大段自然语言文字，而是想要个固定格式，比如"姓名、邮箱、手机号"拆开的好填进数据库。case16.py就干这个。

importosfromlangchain_openaiimportChatOpenAIfrompydanticimportBaseModel,Fieldfromlangchain.agentsimportcreate_agent llm=ChatOpenAI(model='qwen-plus',temperature=0.75,api_key=os.getenv("DASHSCOPE_API_KEY"),base_url="https://ws-89t66lrlxx0rdobw.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",)classContactInfo(BaseModel):name:str=Field(description="姓名")email:str=Field(description="个人邮箱")phone:str=Field(description="个人手机号")agent=create_agent(model=llm,response_format=ContactInfo)

这里用pydantic的BaseModel定义了一个"联系人信息"的结构，Field里写的是字段说明。然后create_agent里传response_format=ContactInfo，等于告诉模型"你就按这个模板给我填，别瞎写"。

跑的时候：

result=agent.invoke({"messages":[{"role":"user","content":"我的联系方式:张三,zhangsan@qq.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 一样 ...checkpointer=InMemorySaver()agent=create_agent(model=llm,tools=tools,system_prompt=system_prompt,checkpointer=checkpointer,debug=True,middleware=[ModelRetryMiddleware(max_retries=3),ToolRetryMiddleware(max_retries=3),ModelCallLimitMiddleware(run_limit=10),SummarizationMiddleware()])

挨个说这几个中间件是干嘛的：

• ModelRetryMiddleware(max_retries=3)：模型调用失败自动重试，最多 3 次。比如网络抖了一下返回 500，它自己重来，不用人管。
• ToolRetryMiddleware(max_retries=3)：工具执行失败也重试，比如调天气接口超时了，重调。
• ModelCallLimitMiddleware(run_limit=10)：限制一轮对话里模型最多被调 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 底层依赖的，不是咱们手写的图）。所以这块我先按老师课上讲的意思给你捋一下，等源码放进来再补。

为啥要 LangGraph？
create_agent好用是真好用，但它把流程"封装死了"——就是个标准的 ReAct 循环，你想改流程（比如"先查数据库，再让两个 agent 分工，最后汇总"）就很不灵活。langchain早期那种链（Chain）又太死板。LangGraph 的思路是把 agent 跑的流程画成一张图（Graph）：节点（Node）是每一步干啥，边（Edge）是走哪条路，还能根据条件分支。这样复杂流程你自己画，想怎么连怎么连。

LangGraph 入门大概长这样（通用写法，项目里暂时没这文件）：

fromlanggraph.graphimportStateGraph,MessagesState,START,ENDfromlanggraph.prebuiltimportToolNode# 1. 定义图，状态用 MessagesState（就是消息列表）builder=StateGraph(MessagesState)# 2. 加节点：一个节点是调模型，一个节点是执行工具builder.add_node("model",call_model)builder.add_node("tools",ToolNode(tools))# 3. 加边：START -> model，tools -> 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. 编译出可执行的图graph=builder.compile()

核心就三样东西：节点（add_node）、普通边（add_edge，按顺序走）、条件边（add_conditional_edges，根据模型输出决定下一步去哪）。把这套学会，比create_agent那种黑盒灵活太多。

等老师把 17-8 的 LangGraph 入门案例代码发下来（或者你自己跟着敲一个case18.py），我再给你按同样的口味讲一遍。