LangChain02 Agent核心双翼

LangChain02 Agent核心双翼

核心目标：掌握Tool Calling（工具调用）和RAG（检索增强生成）。这是让大模型“长出手脚”和“拥有外部大脑”的绝对核心，也是后续学习 LangGraph 的底层基石。
核心心法：理解底层闭环逻辑，拒绝黑盒，手写循环，拥抱 LCEL。

一、Tool Calling：让模型“长出手脚”

核心认知：大模型不会直接执行 Python 代码。它只负责“决定”调用哪个工具并提取参数，真正的执行由后端代码完成。

1. 工具的三要素

使用@tool装饰器定义工具，模型主要依靠以下三要素来决定是否调用：

• name：工具名称。
• description (docstring)：工具能力与适用场景（写得越清晰，调用越准）。
• args schema：参数类型与说明。

2. 核心代码骨架

fromlangchain_core.toolsimporttoolfromlangchain_core.messagesimportHumanMessage,ToolMessage# 1. 定义工具@tooldefget_weather(city:str)->str:"""查询指定城市的天气。当用户询问天气时使用。"""returnf"{city}今天晴天，26度。"tools=[get_weather]# 2. 绑定工具给模型llm_with_tools=llm.bind_tools(tools)# 3. 调用模型并检查 tool_calls (核心！)response=llm_with_tools.invoke([HumanMessage(content="北京天气如何？")])# 判断模型是否决定调用工具ifresponse.tool_calls:fortool_callinresponse.tool_calls:tool_name=tool_call["name"]tool_args=tool_call["args"]tool_id=tool_call["id"]# 4. 后端执行工具result=get_weather.invoke(tool_args)# 5. 包装为 ToolMessage 并喂回模型 (必须带 tool_call_id)tool_msg=ToolMessage(content=result,tool_call_id=tool_id)# 6. 再次调用模型生成最终自然语言回答final_response=llm_with_tools.invoke([HumanMessage(...),response,tool_msg])

3. 手写 Agent 闭环 (LangGraph ToolNode 的底层原理)

用户提问 -> LLM 判断 -> 有 tool_calls? ├── Yes -> 执行 Python 函数 -> 构造 ToolMessage -> 追加到 messages -> 回到 LLM └── No -> 输出最终回答 (response.content)

4.最终代码

importsysfromlangchain_core.messagesimportHumanMessage,ToolMessagefrompathlibimportPath# 将项目根目录加入 sys.path，以便导入 day01 目录中的模块sys.path.insert(0,str(Path(__file__).resolve().parent.parent))fromlangchain_core.toolsimporttoolfromday01.day01importllm@tooldefget_weather(city:str)->str:""" 查询指定城市的天气。 当用户询问某个城市天气时使用。 参数: city: 城市名称，例如“北京”、“上海”。 """weather_data={"北京":"晴天，气温 26 度","上海":"多云，气温 24 度","广州":"小雨，气温 28 度","焦作":"晴天，气温30度","临沂":"多云，气温20度"}returnweather_data.get(city,f"暂时无法查询{city}的天气")@tooldefcalculate_math(expression:str)->str:""" 计算数学表达式。 当用户需要进行数学计算时使用。 参数: expression: 合法的数学表达式，例如 "2 + 3 * 4"。 """try:# 学习演示用，生产环境不要直接用 evalresult=eval(expression,{"__builtins__":{}},{})returnstr(result)exceptExceptionase:returnf"计算失败：{e}"# # 主程序入口# if __name__ == '__main__':""" 1. 用户提问 2. 模型判断需要工具 3. 模型返回 tool_calls 4. 代码根据 tool name 找到对应工具 5. 代码执行工具，得到结果 6. 把结果包装成 ToolMessage 7. 把 ToolMessage 加回 messages 8. 再次调用模型 9. 模型根据工具结果生成最终回答 """tools=[get_weather,calculate_math]# 2. 建立工具名到工具对象的映射tools_by_name={t.name:tfortintools}# 3. 绑定工具给模型llm_with_tools=llm.bind_tools(tools)# 4. 手写 Agent 循环defrun_agent(question:str,max_steps:int=5)->str:messages=[HumanMessage(content=question)]forstepinrange(max_steps):print(f"\n=== 第{step+1}步：调用模型 ===")ai_message=llm_with_tools.invoke(messages)messages.append(ai_message)print("模型返回 content：",ai_message.content)print("模型返回 tool_calls：",ai_message.tool_calls)# 如果没有工具调用，说明模型已经给出最终回答ifnotai_message.tool_calls:returnai_message.content# 如果有工具调用，就执行工具fortool_callinai_message.tool_calls:tool_name=tool_call["name"]tool_args=tool_call["args"]tool_call_id=tool_call["id"]print(f"模型决定调用工具：{tool_name}")print(f"工具参数：{tool_args}")selected_tool=tools_by_name[tool_name]tool_result=selected_tool.invoke(tool_args)print(f"工具执行结果：{tool_result}")# 把工具结果包装成 ToolMessagemessages.append(ToolMessage(content=str(tool_result),tool_call_id=tool_call_id))print("最新的消息为：{messages}")return"达到最大执行步数，任务未完成。"# 5. 测试answer=run_agent("焦作今天天气怎么样？")print("\n最终回答：",answer)

二、RAG：让模型“拥有外部大脑”

核心认知：将私有知识向量化存储，用户提问时检索最相关的片段（Top-K），塞入 Prompt 让模型基于上下文回答，从而解决幻觉和知识滞后问题。

1. RAG 核心流程与组件

2. 核心难点：RunnableLambda格式化

检索器返回的是[Document, Document]对象列表，必须用RunnableLambda将其转为带编号的字符串，以便省 Token 且支持引用。

fromlangchain_core.runnablesimportRunnableLambdadefformat_docs(docs):return"\n\n".join(f"[{i}]{doc.page_content}"fori,docinenumerate(docs,1))format_runnable=RunnableLambda(format_docs)

3. LCEL 组装 RAG Chain

使用字典和RunnablePassthrough并行传递context和question。

fromlangchain_core.runnablesimportRunnablePassthrough rag_chain=({"context":retriever|format_runnable,# 检索并格式化"question":RunnablePassthrough()# 原样传递用户问题}|rag_prompt# 包含 {context} 和 {question} 的 Prompt|llm)answer=rag_chain.invoke("学生请假流程是什么？")

4. 防幻觉 Prompt 设计 (必加)

你必须只根据上下文回答问题。 如果上下文中没有答案，请回答：根据上下文无法确定。 不要编造信息。回答时请使用引用编号，例如 [1]、[2]。

5.RAG最终代码

importosfromdotenvimportload_dotenvfromlangchain_core.documentsimportDocumentfromlangchain_openaiimportChatOpenAI# from langchain_community.document_loaders import TextLoaderfromlangchain_text_splittersimportRecursiveCharacterTextSplitterfromlangchain_community.embeddingsimportDashScopeEmbeddingsfromlangchain_community.vectorstoresimportFAISSfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.runnablesimportRunnableLambda,RunnablePassthrough load_dotenv()# 1. 初始化 LLMllm=ChatOpenAI(model="kimi-k2.7-code",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",api_key=os.getenv("DASHSCOPE_API_KEY"),temperature=1,)# 2. 初始化 Embeddingsembeddings=DashScopeEmbeddings(model=os.getenv("embedding_name"),dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"))# 3. 加载文档# (1) 用 Python 原生方式读取文件file_path="handbook.txt"withopen(file_path,"r",encoding="utf-8")asf:text=f.read()# (2) 手动包装成 LangChain 的 Document 对象documents=[Document(page_content=text,metadata={"source":file_path}# metadata 用于记录来源，RAG 中很有用)]# 4. 切分文档splitter=RecursiveCharacterTextSplitter(chunk_size=300,chunk_overlap=60)chunks=splitter.split_documents(documents)print(f"切分文档后chunks的类型是{type(chunks)}")# 5. 存入向量数据库vectorstore=FAISS.from_documents(documents=chunks,embedding=embeddings)# 6. 创建检索器retriever=vectorstore.as_retriever(search_type="similarity",search_kwargs={"k":3})# 7. 格式化检索结果defformat_docs(docs):return"\n\n".join(f"[{i}]{doc.page_content.strip()}"fori,docinenumerate(docs,1))# 8. 定义 Promptrag_prompt=ChatPromptTemplate.from_messages([("system",""" 你是一个本地文档问答助手。 你必须只根据上下文回答问题。 如果上下文中没有答案，请回答：根据上下文无法确定。 不要编造信息。 回答时请使用引用编号，例如 [1]、[2]。 上下文： {context} """),("human","{question}")])# 9. 构建 RAG Chainrag_chain=({"context":retriever|RunnableLambda(format_docs),"question":RunnablePassthrough()}|rag_prompt|llm)# 10. 测试question="硕士1班有什么学生，学号是多少"response=rag_chain.invoke(question)print(response.content)

三、避坑指南 (Day 2 Don’ts)

• ❌不要看response.content来判断工具调用：必须看response.tool_calls。
• ❌不要漏掉tool_call_id：ToolMessage必须带上 ID，否则模型不知道结果对应哪次调用。
• ❌不要用老版TextLoader：会报 DeprecationWarning，直接用 Pythonopen()+Document。
• ❌不要学老版RetrievalQA：直接用 LCEL 自己拼装 RAG Chain。
• ❌不要碰复杂的 Memory 类：如ConversationBufferMemory，LangGraph 中用messages状态和Checkpointer管理记忆。
• ❌不要一上来学高级 RAG：先跑通基础 Top-K 检索，不要碰 Parent Document、Hybrid Search 等。

四、面试Q&A

Q1：Function Calling 的本质是什么？模型真的执行代码了吗？

A:模型没有执行代码。本质是模型根据用户输入和工具的 description/schema，生成一个结构化的 JSON 请求（包含工具名、参数和 ID）。后端代码解析这个请求，执行真正的 Python 函数，然后将结果通过ToolMessage喂回给模型，模型再基于结果生成自然语言回答。

Q2：为什么 RAG 中需要chunk_overlap（文本重叠）？

A:因为机械切分文本可能会把完整的一句话或一个逻辑段落从中间切断，导致语义丢失。设置chunk_overlap让相邻的文本块保留一部分重叠内容，可以保证上下文语义的连贯性，提高检索准确率。

Q3：在 LCEL 中构建 RAG 时，为什么要用RunnableLambda？

A:因为 Retriever 检索出来的是Document对象列表，而 Prompt 需要的是纯文本字符串。RunnableLambda允许我们将自定义的 Python 格式化函数（如拼接带[1] [2]编号的字符串）包装成 LCEL 兼容的组件，无缝接入管道|中。

Q4：手写的 Tool Calling while 循环和 LangGraph 有什么关系？

A:手写的 while 循环就是 LangGraph 中 Agent 节点和 Tool 节点交互的底层逻辑。在 LangGraph 中，这个循环被抽象成了状态图：Agent 节点判断是否有tool_calls，如果有，通过条件边（Conditional Edge）路由到 Tool 节点执行，执行完后再路由回 Agent 节点，直到没有工具调用为止。