LangChain学习笔记（一）

智能体

AI Agent（智能代理）是指一个能够感知环境、进行推理决策、并在环境中执行行动以实现特定目标的智能系统。它具有以下核心特征：

1. 自主性（Autonomy）：能够在没有人类直接干预的情况下独立运行和做出决策
2. 反应性（Reactivity）：能够感知环境的变化并及时做出响应
3. 主动性（Proactivity）：不仅能被动响应，还能主动采取行动来实现目标
4. 社交能力（Social Ability）：能够与其他Agent或人类进行交互和协作

AI Agent 深度学习笔记：概念辨析与设计模式详解

创建模型

这里创建模型有两种方式：

方式一：

from langchain.chat_models import init_chat_model import os os.environ["OPENAI_API_KEY"] = "sk-..." model = init_chat_model("openai:gpt-4.1", temperature=0.7, max_tokens=500) response = model.invoke("给我讲一个关于AI的笑话") print(response)

方式二：

from dotenv import load_dotenv from langchain_anthropic import ChatAnthropic # 加载 .env 文件中的变量，在项目目录下创建一个.env文件，把环境变量配置进去，入API_KEY和BASE_URL # ANTHROPIC_API_KEY=sk-...... # ANTHROPIC_API_BASE_URL=https://api.moonshot.cn/anthropic load_dotenv() #kimi是兼容Anthropic的，所以我以kimi为例创建 model = ChatAnthropic( model="kimi-k2.6", temperature=0.1, max_tokens=8000, streaming=True )

文档教程中分静态模型和动态模型，两者的区别是：

• 静态模型在创建智能体时配置一次，并在整个执行过程中保持不变。这是最常见且直接的方法。
• 动态模型在 运行时 根据当前 状态 和上下文进行选择。这支持复杂的路由逻辑和成本优化。要使用动态模型，请使用 @wrap_model_call 装饰器创建中间件，以修改请求中的模型：

from langchain_openai import ChatOpenAI from langchain.agents import create_agent from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse basic_model = ChatOpenAI(model="gpt-4o-mini") advanced_model = ChatOpenAI(model="gpt-4o") @wrap_model_call def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse: """根据对话复杂性选择模型。""" message_count = len(request.state["messages"]) if message_count > 10: # 对较长的对话使用高级模型 model = advanced_model else: model = basic_model request.model = model return handler(request) agent = create_agent( model=basic_model, # 默认模型 tools=tools, middleware=[dynamic_model_selection] )

创建工具

工具是代理（agents）调用来执行操作的组件。它们通过允许模型通过定义明确的输入和输出与世界交互来扩展模型的功能。工具封装了一个可调用的函数及其输入架构（schema）。这些可以传递给兼容的聊天模型（chat models），让模型决定是否以及使用什么参数来调用工具。在这些场景中，工具调用使模型能够生成符合指定输入架构的请求。

创建工具最简单的方法是使用@tool装饰器。默认情况下，函数的文档字符串（docstring）会成为工具的描述，帮助模型理解何时使用它：

from langchain.tools import tool from langchain.agents import create_agent @tool def search(query: str) -> str: """搜索信息。""" return f"结果：{query}" @tool def get_weather(location: str) -> str: """获取位置的天气信息。""" return f"{location} 的天气：晴朗，72°F" agent = create_agent(model, tools=[search, get_weather])

工具错误处理

要自定义工具错误的处理方式，请使用 @wrap_tool_call 装饰器创建中间件：

from langchain_anthropic import ChatAnthropic from langchain.agents import create_agent from typing import TypedDict from dotenv import load_dotenv from langchain.agents.middleware import wrap_tool_call, dynamic_prompt, ModelRequest from langchain_core.messages import ToolMessage, AIMessageChunk, AIMessage from langchain.agents import AgentState from dataclasses import dataclass from langchain.tools import tool, ToolRuntime load_dotenv() @tool def get_weather_for_location(city: str) -> str: """获取指定城市的天气。""" return f"{city}总是阳光明媚！" @dataclass class Context: """自定义运行时上下文模式。""" user_id: str @tool def get_user_location(runtime: ToolRuntime[Context]) -> str: """根据用户 ID 获取用户信息。""" user_id = runtime.context.user_id return "Florida" if user_id == "1" else "SF"# 加载 .env 文件中的变量 model = ChatAnthropic( model="kimi-k2.6", temperature=0.1, max_tokens=8000, streaming=True ) # 工具错误处理 @wrap_tool_call def handle_tool_errors(request, handler): """使用自定义消息处理工具执行错误。""" try: return handler(request) except Exception as e: # 向模型返回自定义错误消息 return ToolMessage( content=f"工具错误：请检查您的输入并重试。({str(e)})", tool_call_id=request.tool_call["id"] ) SYSTEM_PROMPT = """你是一位擅长用双关语表达的专家天气预报员。 你可以使用两个工具： - get_weather_for_location：用于获取特定地点的天气 - get_user_location：用于获取用户的位置 如果用户询问天气，请确保你知道具体位置。如果从问题中可以判断他们指的是自己所在的位置，请使用 get_user_location 工具来查找他们的位置。""" agent = create_agent( model, system_prompt=SYSTEM_PROMPT, middleware=[handle_tool_errors], tools=[get_weather_for_location, get_user_location], )

系统提示词（System Prompt）

系统提示词是在用户输入之前，预先设置给AI模型的指令或背景信息，用于引导模型的行为、风格和能力范围。

📌 核心概念

什么是系统提示词？

系统提示词就像是给AI的"角色设定卡"或"工作说明书"，它定义了：

• AI的身份和角色
• 应该遵循的规则
• 输出的风格和格式
• 能力边界和限制

⚠️ 位置和时机

┌─────────────────────────────────┐ │ 系统提示词（System Prompt） │ ← 预先设定，用户不可见 ├─────────────────────────────────┤ │ 用户输入（User Message） │ ← 用户可见 ├─────────────────────────────────┤ │ AI回复（Assistant Response） │ ← AI生成 └─────────────────────────────────┘

🎯 系统提示词的作用

💡 优秀系统提示词的特点

1.明确具体

# ❌ 模糊 "你是一个有帮助的助手" # ✅ 具体 "你是一位专业的天气预报员，擅长用双关语表达天气信息。只回答与天气相关的问题，如果用户问其他问题，请礼貌拒绝。"

2.结构清晰

SYSTEM_PROMPT = """ # 角色 你是一位资深的Python开发工程师 # 能力 - 解答Python编程问题 - 提供代码优化建议 - 解释技术概念 # 规则 1. 代码必须符合PEP 8规范 2. 优先使用标准库 3. 提供详细注释 # 输出格式 - 代码块使用```python标记 - 复杂概念分步骤解释 """

3.包含示例（Few-shot）

SYSTEM_PROMPT = """ 你是一个情感分析助手。 示例： 用户："这个产品太棒了！" 输出：{"情感": "正面", "置信度": 0.95} 用户："质量很差，不推荐购买" 输出：{"情感": "负面", "置信度": 0.88} 请按照上述格式分析用户输入的情感。 """

🛠️ 实际应用示例

1. 代码审查助手

SYSTEM_PROMPT = """ 你是一位严格的代码审查专家。 # 审查要点 1. 代码规范性（PEP 8） 2. 潜在的bug 3. 性能问题 4. 安全隐患 5. 可读性 # 输出格式 按严重程度分类： - 🔴 严重问题 - 🟡 中等问题 - 🟢 建议优化 每个问题包含：问题描述 + 修改建议 """

2. 常见误区

3. 系统提示词与用户提示词的区别

🔧 使用方式

1. system_prompt

SYSTEM_PROMPT = """你是一位擅长用双关语表达的专家天气预报员。 你可以使用两个工具： - get_weather_for_location：用于获取特定地点的天气 - get_user_location：用于获取用户的位置 如果用户询问天气，请确保你知道具体位置。如果从问题中可以判断他们指的是自己所在的位置，请使用 get_user_location 工具来查找他们的位置。""" agent = create_agent( model, system_prompt=SYSTEM_PROMPT, )

2.动态系统提示词

class Context(TypedDict): user_role: str @dynamic_prompt def user_role_prompt(request: ModelRequest) -> str: user_role = request.runtime.context.get("user_role", "user") if user_role == "expert": return "提供详细的技术响应。" elif user_role == "beginner": return "简单解释概念，避免使用行话。" return "你是一个有帮助的助手。" agent = create_agent( model, system_prompt=SYSTEM_PROMPT, middleware=[handle_tool_errors, user_role_prompt], tools=[get_weather_for_location, get_user_location], context_schema=Context, ) # 系统提示将根据上下文动态设置 result = agent.invoke( { "messages": [{"role": "user", "content": "什么是量化交易"}], }, context={"user_role": "expert"}, ) print(result)

🎯 总结

系统提示词是控制AI行为最强大、最直接的工具。好的系统提示词应该：

✅明确具体- 清晰定义角色和规则
✅结构清晰- 逻辑层次分明
✅包含示例- 提供few-shot学习样本
✅持续优化- 根据实际效果迭代改进

调用

result = agent.invoke( {"messages": [{"role": "user", "content": "旧金山天气如何？"}]} )

输出

结构化输出

控制智能体以特定格式返回输出：ToolStrategy使用人工工具调用生成结构化输出。这适用于任何支持工具调用的模型：

from pydantic import BaseModel from langchain.agents import create_agent from langchain.agents.structured_output import ToolStrategy class ContactInfo(BaseModel): name: str email: str phone: str agent = create_agent( model="kimi-k2.6", response_format=ToolStrategy(ContactInfo) ) result = agent.invoke({ "messages": [{"role": "user", "content": "从以下内容提取联系信息：John Doe, john@example.com, (555) 123-4567"}] }) result["structured_response"] # ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')

ProviderStrategy使用模型提供商的原生结构化输出生成。这更可靠，但仅适用于支持原生结构化输出的提供商（例如 OpenAI）：

from langchain.agents.structured_output import ProviderStrategy agent = create_agent( model="openai:gpt-4o", response_format=ProviderStrategy(ContactInfo) )

流式输出

流式输出让 AI 的回复像打字一样逐字显示，极大地提升了用户体验。LangChain 的 Agent 内置了完善的流式输出支持。

为什么需要流式输出

如果使用 invoke()，用户需要等待 Agent 完成所有步骤（多次模型调用 + 工具执行）才能看到结果。对于复杂任务，这可能耗时十几秒甚至更长。流式输出解决了这个问题：每生成一个 Token 就立即返回，用户可以实时看到进展。

stream_mode="messages"

这是最细粒度的流式模式，每个 chunk 对应一个 Token——逐 Token 流式：

from dotenv import load_dotenv load_dotenv() from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.messages import HumanMessage model = init_chat_model(model="kimi-k2.6", model_provider="anthropic") agent = create_agent( model=model, system_prompt="你是互联网的百事通。", ) # stream_mode="messages" 逐个 Token 返回 print("实时流式输出：") for msg_chunk, metadata in agent.stream( {"messages": [HumanMessage(content="阐述一下什么是AI Agent")]}, stream_mode="messages", ): # msg_chunk 是 AIMessageChunk # 每个 chunk 只包含一小段内容 if msg_chunk.content: print(msg_chunk.content, end="", flush=True) print() # 最后换行

metadata 包含了这个 chunk 的来源信息：

print("查看 metadata 信息：\n") for msg_chunk, metadata in agent.stream( {"messages": [HumanMessage(content="你好，介绍一下你自己")]}, stream_mode="messages", ): if msg_chunk.content and len(msg_chunk.content) > 5: print(f"内容: {msg_chunk.content}") print(f"来源节点: {metadata.get('langgraph_node')}") print(f"消息类型: {type(msg_chunk).__name__}") break # 只看第一个有意义的 chunk # 输出: 查看 metadata 信息： 内容: 你好！我是菜 来源节点: model 消息类型: AIMessageChunk

stream_mode="updates"

这个模式在构建需要显示"思考过程"的界面时非常有用——逐步查看 Agent 执行过程：

from langchain.tools import tool @tool def search_course(keyword: str) -> str: """在菜鸟教程 RUNOOB 搜索课程""" courses = { "python": "Python3 基础教程（30章，20小时）", "html": "HTML 基础教程（25章，15小时）", } return courses.get(keyword.lower(), "未找到相关课程") agent = create_agent( model = init_chat_model(model="kimi-k2.6", model_provider="anthropic"), tools=[search_course], system_prompt="你是菜鸟教程 RUNOOB 的课程顾问。", ) # 使用 updates 模式查看每一步 print("=== Agent 执行过程 ===\n") for chunk in agent.stream( {"messages": [HumanMessage(content="帮我查一下 Python 课程")]}, stream_mode="updates", ): for node_name, update in chunk.items(): print(f"[{node_name}]", end=" ") if "messages" in update: for msg in update["messages"]: if msg.type == "ai": if hasattr(msg, 'tool_calls') and msg.tool_calls: calls = [tc['name'] for tc in msg.tool_calls] print(f"请求调用: {calls}") elif msg.content: print(f"回复: {msg.content[:80]}") elif msg.type == "tool": print(f"工具返回 [{msg.name}]: {msg.content}")

stream_mode="custom"

通过 Middleware 的 runtime.stream_writer()，你可以向流中发送自定义事件：

from langchain.agents.middleware import before_model, after_model @before_model def notify_before(state, runtime): """在模型调用前发送自定义事件""" runtime.stream_writer({ "type": "status", "message": "正在思考...", }) return None @after_model def notify_after(state, runtime): """在模型调用后发送自定义事件""" last_msg = state["messages"][-1] if state.get("messages") else None has_tools = hasattr(last_msg, 'tool_calls') and last_msg.tool_calls if has_tools: tool_names = [tc['name'] for tc in last_msg.tool_calls] runtime.stream_writer({ "type": "status", "message": f"正在调用工具: {', '.join(tool_names)}...", }) else: runtime.stream_writer({ "type": "status", "message": "回答已完成", }) return None agent = create_agent( model = init_chat_model(model="kimi-k2.6", model_provider="anthropic"), tools=[search_course], middleware=[notify_before, notify_after], system_prompt="你是菜鸟教程 RUNOOB 的课程顾问。", ) # 使用 stream_mode=["updates", "custom"] 同时接收两种事件 print("=== 混合流式输出 ===\n") for mode, chunk in agent.stream( {"messages": [HumanMessage(content="查一下 Python 课程")]}, stream_mode=["updates", "custom"], ): if mode == "custom": print(f"[自定义事件] 状态: {chunk['message']}") elif mode == "updates": for node_name, update in chunk.items(): if "messages" in update: for msg in update["messages"]: if msg.type == "ai" and msg.content: print(f"[回复] {msg.content}")

stream_mode 可以组合使用，如 stream_mode=["updates", "custom", "messages"]。但过多的模式会增加流中的事件量，建议按需选择。

异步流式输出

在 Web 服务中，使用异步流式可以避免阻塞事件循环：

import asyncio async def stream_agent(): """异步流式运行 Agent""" agent = create_agent( model=init_chat_model("deepseek:deepseek-v4-flash"), system_prompt="你是菜鸟教程 RUNOOB 的助手。", ) full_response = "" async for msg_chunk, metadata in agent.astream( {"messages": [HumanMessage(content="一句话介绍菜鸟教程")]}, stream_mode="messages", ): if msg_chunk.content: full_response += msg_chunk.content print(msg_chunk.content, end="", flush=True) print(f"\n\n完整回复长度: {len(full_response)} 字") # 运行异步函数 asyncio.run(stream_agent())

在Django中使用

# views.py from django.http import StreamingHttpResponse from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.messages import HumanMessage # ⚠️ 注意：不再使用 @api_view，改用 Django 原生 async 视图 # 如果你必须用 DRF，请升级到 DRF 3.15+ 并使用 @api_view 的 async 支持 async def chat_stream(request): """纯异步流式聊天接口""" message = request.GET.get("message", "一句话介绍菜鸟教程") # ✅ 每次请求独立创建 agent，避免状态污染 agent = create_agent( model=init_chat_model("deepseek:deepseek-v4-flash"), system_prompt="你是菜鸟教程 RUNOOB 的助手。", ) async def generate(): async for msg_chunk, metadata in agent.astream( {"messages": [HumanMessage(content=message)]}, stream_mode="messages", ): if msg_chunk.content: yield f"data: {msg_chunk.content}\n\n" yield "data: [DONE]\n\n" return StreamingHttpResponse( generate(), content_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # Nginx 反代时必须加 }, ) # urls.py from django.urls import path from . import views urlpatterns = [ path("chat/", views.chat_stream), ]

必须用 ASGI 服务器运行（如 Uvicorn/Daphne），WSGI 服务器不支持异步视图。

uvicorn myproject.asgi:application --host 0.0.0.0 --port 8000

Nginx 反向代理配置

location /chat/ { proxy_pass http://127.0.0.1:8000; proxy_buffering off; # ← 关键！否则SSE会被缓冲 proxy_cache off; chunked_transfer_encoding on; proxy_set_header Connection ''; proxy_http_version 1.1; }

中间件

中间件为在执行的不同阶段自定义智能体行为提供了强大的扩展性。您可以使用中间件来：

• 在调用模型之前处理状态（例如消息裁剪、上下文注入）
• 修改或验证模型的响应（例如防护栏、内容过滤）
• 使用自定义逻辑处理工具执行错误
• 基于状态或上下文实现动态模型选择
• 添加自定义日志、监控或分析

中间件无缝集成到智能体的执行图中，允许您在关键点拦截和修改数据流，而无需更改核心智能体逻辑。使用方法是通过将其传递给create_agent函数来添加中间件：

from langchain.agents import create_agent from langchain.agents.middleware import SummarizationMiddleware, HumanInTheLoopMiddleware agent = create_agent( model="kimi-k2.6", tools=[...], middleware=[SummarizationMiddleware(), HumanInTheLoopMiddleware()], )