快速上手 LangChain v1 教程(上):核心概念与基础构建
本教程基于 LangChain v1 官方文档(https://langchain-doc.cn/v1/python/)整理,适用于想系统掌握 LangChain 构建智能体(Agent)的开发者。本篇为上半部分,涵盖 LangChain 的核心抽象、模型调用、消息系统、工具集成、记忆机制与结构化输出等。
一、简介:LangChain 是什么?
LangChain 是一个用于构建 AI 智能体(Agents) 的 Python 框架,其核心目标是:
- 将 LLM(大语言模型)与 外部工具 结合
- 提供 记忆(Memory) 能力以维持上下文
- 支持 结构化输出 供下游系统直接使用
- 通过 中间件(Middleware) 实现执行流程的精细控制
LangChain v1 基于 LangGraph 构建,将 Agent 表达为 状态图(State Graph),实现可追踪、可调试、可持久化的执行流程。
二、安装与初始化
1. 安装基础包
pip install langchain langchain-openai langchain-anthropic langgraph
LangChain 采用模块化设计,各模型提供商(如 OpenAI、Anthropic)需单独安装对应集成包。
2. 初始化模型
LangChain 支持两种方式指定模型:
字符串标识(自动推断)
from langchain.agents import create_agent
agent = create_agent("openai:gpt-4o", tools=[])
手动初始化(高定制)
# 第一种
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
model = ChatOpenAI(
model="gpt-4o",
temperature=0.1,
max_tokens=1000,
timeout=30,
)
agent = create_agent(model, tools=[])
# 第二种
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
open_model = init_chat_model(
model = "THUDM/GLM-Z1-9B-0414",
model_provider = "openai",
api_key = "sk-***************************",
base_url = "https://api.siliconflow.cn/v1/",
)
agent = create_agent(
model=open_model,
system_prompt="""你是一名人工助手""",
tools=[search, get_weather]
)
3. 代码示例 (以硅基流动API为例子)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
if __name__ == "__main__":
open_model = init_chat_model(
model = "THUDM/GLM-Z1-9B-0414",
model_provider = "openai",
api_key = "sk-*********************",
base_url = "https://api.siliconflow.cn/v1/",
)
agent = create_agent(
model=open_model,
system_prompt = """你是一名人工助手,专门帮助用户完成任务与处理问题""",
)
result = agent.invoke({"messages": [{"role": "user", "content": "如何创建一个txt文件"}]})
print(result.get("messages")[1].content)
✅ 最佳实践:建议使用
ChatOpenAI、ChatAnthropic、init_chat_model等实例化方式,便于精细控制模型参数。
三、消息系统(Messages)
LangChain 中,所有输入/输出均以 消息对象 表示,这是与 LLM 交互的标准单位。
1. 消息类型
| 类型 | 用途 | 示例 |
|---|---|---|
SystemMessage |
设定角色、行为规范 | “你是一个 Python 专家” |
HumanMessage |
用户输入 | “如何创建 REST API?” |
AIMessage |
模型输出(含工具调用) | 模型回复或 tool_calls |
ToolMessage |
工具执行结果 | 工具返回值 + tool_call_id |
2. 创建消息
from langchain.messages import SystemMessage, HumanMessage
messages = [
SystemMessage("你是一个乐于助人的助手。"),
HumanMessage("你好!")
]
也可以使用字典格式(兼容 OpenAI API):
messages = [
{"role": "system", "content": "你是一个乐于助人的助手。"},
{"role": "user", "content": "你好!"}
]
3. 消息属性
content:原始内容(字符串或列表)tool_calls:模型请求的工具调用列表usage_metadata:输入/输出 token 数content_blocks:标准化多模态内容块(v1 新特性)
4. 代码示例 (以硅基流动API为例子)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, AIMessage, SystemMessage
if __name__ == "__main__":
open_model = init_chat_model(
model = "THUDM/GLM-Z1-9B-0414",
model_provider = "openai",
api_key = "sk-*********************v",
base_url = "https://api.siliconflow.cn/v1/",
)
human_msg = HumanMessage("如何创建一个txt文件")
system_msg = SystemMessage("你是一名人工助手,专门帮助用户完成任务与处理问题")
messages = [human_msg, system_msg]
result = open_model.invoke(messages)
print(result.content)
四、工具(Tools):赋予 Agent 行动力
工具是 Agent 执行外部操作的能力扩展,如搜索、数据库查询、API 调用等。
1. 定义工具
使用 @tool 装饰器将函数转为工具:
from langchain.tools import tool
@tool
def get_weather(location: str) -> str:
"""Get the weather at a location."""
return f"{location} is sunny."
2. 注册工具
传递工具列表给 create_agent:
agent = create_agent(model, tools=[get_weather])
⚠️ 若工具列表为空,Agent 将退化为纯 LLM 调用,无法调用工具。
3. 工具错误处理
使用中间件拦截工具异常:
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
@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"]
)
4. 代码示例 (以硅基流动API为例子)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
@tool
def search(query: str) -> str:
"""搜索信息。"""
return f"结果:{query}"
@tool
def get_weather(location: str) -> str:
"""获取位置的天气信息。"""
return f"{location} 的天气:晴朗,36°C"
if __name__ == "__main__":
open_model = init_chat_model(
model = "THUDM/GLM-Z1-9B-0414",
model_provider = "openai",
api_key = "sk-*********************",
base_url = "https://api.siliconflow.cn/v1/",
)
agent = create_agent(
model=open_model,
system_prompt="""你是一名人工助手""",
tools=[search, get_weather]
)
result = agent.invoke({"messages": [{"role": "user", "content": "告诉我北京的天气,并搜索相关新闻"}]})
print(result)
五、记忆(Short-Term Memory):维持对话上下文
LangChain Agent 通过 状态(State) 管理短期记忆,默认使用 AgentState,其中 messages 键存储对话历史。
1. 启用持久化记忆
使用 checkpointer 实现线程级持久化:
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(
model,
tools=[],
checkpointer=InMemorySaver()
)
# 指定 thread_id 实现会话隔离
agent.invoke(
{"messages": [{"role": "user", "content": "我是 Bob"}]},
{"configurable": {"thread_id": "1"}}
)
📌 线程(Thread) 相当于一个独立会话,类似邮件对话线程。
2. 自定义状态(扩展记忆)
from langchain.agents import AgentState
from typing_extensions import TypedDict
class CustomState(AgentState):
user_id: str
preferences: dict
agent = create_agent(
model,
tools=[],
state_schema=CustomState
)
result = agent.invoke({
"messages": [{"role": "user", "content": "你好"}],
"user_id": "user_123",
"preferences": {"theme": "dark"}
})
✅ v1 要求:自定义状态必须是
TypedDict,不再支持 Pydantic 模型。
3. 管理长对话(避免上下文溢出)
LLM 上下文窗口有限,长对话需主动管理:
| 策略 | 说明 |
|---|---|
| 修剪(Trim) | 保留最近 N 条消息 |
| 删除(Delete) | 永久移除旧消息 |
| 摘要(Summarize) | 用 LLM 生成历史摘要 |
| 自定义 | 如过滤无关消息 |
示例:自动摘要(SummarizationMiddleware)
from langchain.agents.middleware import SummarizationMiddleware
agent = create_agent(
model="openai:gpt-4o",
tools=[],
middleware=[
SummarizationMiddleware(
model="gpt-4o-mini",
max_tokens_before_summary=4000,
messages_to_keep=20
)
]
)
4. 代码示例 (以硅基流动API为例子)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
@tool
def search(query: str) -> str:
"""搜索信息。"""
return f"结果:{query}"
@tool
def get_weather(location: str) -> str:
"""获取位置的天气信息。"""
return f"{location} 的天气:晴朗,72°F"
if __name__ == "__main__":
open_model = init_chat_model(
model = "THUDM/GLM-Z1-9B-0414",
model_provider = "openai",
api_key = "sk-*********************",
base_url = "https://api.siliconflow.cn/v1/",
)
agent = create_agent(
model=open_model,
system_prompt="""你是一名人工助手""",
tools=[search, get_weather],
checkpointer=InMemorySaver(),
)
result = agent.invoke({"messages": [{"role": "user", "content": "告诉我北京的天气,并搜索相关新闻"}]},
{"configurable": {"thread_id": "1"}})
print(result)
六、结构化输出(Structured Output)
让 Agent 以 JSON / Pydantic / TypedDict 等格式返回数据,避免解析自然语言。
1. 两种策略
| 策略 | 适用模型 | 可靠性 |
|---|---|---|
ProviderStrategy |
OpenAI、Grok(原生支持) | ⭐⭐⭐⭐⭐ |
ToolStrategy |
所有支持工具调用的模型 | ⭐⭐⭐⭐ |
⚠️ v1 注意:不再支持
response_format=MyModel,必须显式使用ToolStrategy或ProviderStrategy。
2. 示例:提取联系人信息
from pydantic import BaseModel, Field
from langchain.agents.structured_output import ToolStrategy
class ContactInfo(BaseModel):
name: str = Field(description="姓名")
email: str = Field(description="邮箱")
phone: str = Field(description="电话")
agent = create_agent(
model="openai:gpt-4o-mini",
tools=[],
response_format=ToolStrategy(ContactInfo)
)
result = agent.invoke({
"messages": [{"role": "user", "content": "John Doe, john@example.com, (555)123-4567"}]
})
print(result["structured_response"])
# ContactInfo(name='John Doe', email='john@example.com', phone='(555)123-4567')
3. 错误处理
ToolStrategy 支持自动重试与自定义错误消息:
ToolStrategy(
schema=ContactInfo,
handle_errors="请提供完整的姓名、邮箱和电话。"
)
上半部分总结:我们已掌握 LangChain v1 的核心抽象——模型、消息、工具、记忆与结构化输出。这些是构建智能体的基石。
(下半部分预告:中间件、流式传输、动态提示、多模态支持、生产部署最佳实践)
更多推荐


所有评论(0)