LangChain create_agent() 函数
create_agent() 是 LangChain 1.x 中创建 AI Agent 的核心函数。
它可以将语言模型、工具、系统提示词和中间件组合起来,创建一个能够自动调用工具并循环执行任务的 Agent。
与普通模型调用不同,Agent 不仅可以回答问题,还能根据任务需要调用 Python 函数、读取外部数据,并根据工具返回的结果决定下一步操作。
在 LangChain 1.x 中,create_agent() 基于 LangGraph 构建并编译 Agent 执行图,支持状态管理、流式输出、结构化响应、运行时上下文和会话持久化等功能。
本章节将完整介绍 create_agent() 的参数和常用方法,并使用 DeepSeek Flash 模型进行演示。
一、create_agent() 的工作原理
使用 create_agent() 创建的 Agent,通常按照以下流程运行:
用户输入
|
v
Model 节点
|
| 模型是否发起工具调用?
|
+-- 否 --> 返回最终结果
|
+-- 是
|
v
Tools 节点
|
| 执行工具
| 将结果加入消息历史
|
v
Model 节点
|
| 继续判断是否调用工具
v
...
其中:
- Model 节点:负责调用语言模型,生成回复或工具调用请求。
- Tools 节点:负责执行模型请求的客户端工具,并返回工具结果。
- Agent State:保存当前消息及其他状态字段。
- Middleware:可以在执行过程中的特定阶段修改或控制 Agent 行为。
Agent 一般会持续执行,直到模型不再请求工具,或者触发其他停止条件。
因此,create_agent() 创建的并不是单纯的聊天模型,而是一个具有执行流程和状态管理能力的应用组件。
二、安装与环境准备
本教程使用 LangChain 1.x,并以 DeepSeek 为主要模型提供商。
安装依赖:
pip install -U langchain langchain-deepseek langgraph python-dotenv pydantic
在当前项目目录创建 .env 文件:
DEEPSEEK_API_KEY=your_deepseek_api_key
Python 程序通过 load_dotenv() 加载密钥。
创建第一个 Agent
实例
from langchain.agents import create_agent
load_dotenv()
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
system_prompt="你是一名中文编程助手。"
)
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "什么是 Python?"
}
]
})
print(result["messages"][-1].content)
这里没有给 Agent 配置任何工具,因此它主要通过模型生成回答。
虽然没有工具,create_agent() 依然会返回一个可执行的 LangGraph 对象。
注意:DeepSeek 当前推荐的模型名称包括 deepseek-flash 和 deepseek-v4-pro。新教程不再使用旧的 deepseek-v4-flash 作为默认模型名称。
三、create_agent() 函数语法
根据 LangChain 官方 API Reference,函数语法如下:
from langchain.agents import create_agent
agent = create_agent(
model,
tools=None,
*,
system_prompt=None,
middleware=(),
response_format=None,
state_schema=None,
context_schema=None,
checkpointer=None,
store=None,
interrupt_before=None,
interrupt_after=None,
debug=False,
name=None,
cache=None,
transformers=None,
)
参数说明:
| 参数 | 说明 | 默认值 |
|---|---|---|
model |
模型名称字符串或 BaseChatModel 实例 | 必填 |
tools |
工具列表,支持可执行工具、函数和受支持的提供商工具定义 | None |
system_prompt |
系统提示词,可以是字符串或 SystemMessage | None |
middleware |
Agent 中间件列表 | () |
response_format |
结构化输出 Schema 或策略 | None |
state_schema |
扩展 AgentState 的自定义状态结构 | None |
context_schema |
运行时 Context 结构 | None |
checkpointer |
保存和恢复会话状态的检查点组件 | None |
store |
用于跨会话共享数据的 Store | None |
interrupt_before |
指定在哪些图节点执行前暂停 | None |
interrupt_after |
指定在哪些图节点执行后暂停 | None |
debug |
是否输出图执行调试信息 | False |
name |
编译后 Agent 图的名称 | None |
cache |
LangGraph 图执行缓存配置 | None |
transformers |
为图注册额外的流事件转换器 | None |
对于初学者,最常用的是 model、tools 和 system_prompt。
构建复杂 Agent 时,再逐步使用 middleware、response_format、checkpointer 等高级参数。
四、model 参数:指定语言模型
model 是必填参数,主要支持两种形式:
- 模型名称字符串,例如
deepseek:deepseek-flash。 - 已经创建好的
BaseChatModel实例。
方式一:直接传入模型名称
实例
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
system_prompt="你是菜鸟教程 RUNOOB 的编程助手。"
)
这种写法简单直接,LangChain 会根据模型名称选择对应提供商并初始化模型。
推荐明确使用:
provider:model
例如:
deepseek:deepseek-flash deepseek:deepseek-v4-pro openai:gpt-4o-mini
实际使用时,需要安装对应提供商集成包并配置 API Key。
方式二:传入已初始化的模型
实例
from langchain.chat_models import init_chat_model
# 初始化模型并配置参数
model = init_chat_model(
"deepseek:deepseek-flash",
max_tokens=500,
timeout=30,
max_retries=2
)
agent = create_agent(
model=model,
tools=[],
system_prompt="你是菜鸟教程 RUNOOB 的编程助手。"
)
这种方式适合需要配置模型超时、重试次数、输出上限或复用模型实例的场景。
注意:不推荐在传给 create_agent() 之前自行调用 model.bind_tools()。
因为 Agent 会根据其工具列表和运行时策略管理工具绑定。预先绑定工具可能与 Agent 的工具注册或动态工具管理机制发生冲突。
五、tools 参数:注册 Agent 工具
tools 用于指定 Agent 可以调用哪些工具。
官方支持的主要类型包括:
BaseTool:LangChain 可执行工具对象。- 普通 Python Callable:可以被转换为工具。
- 字典形式的提供商原生工具定义:由相应模型服务处理。
需要特别区分:工具的参数 Schema 不等于工具的实际执行逻辑。
方式一:使用 @tool 装饰器
实例
from langchain.agents import create_agent
@tool
def search_course(keyword: str) -> str:
"""根据关键词搜索模拟编程课程。"""
courses = {
"python": "Python 基础、Python 数据分析",
"java": "Java 基础、Java 面向对象"
}
return courses.get(
keyword.lower(),
"未找到相关模拟课程"
)
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[search_course],
system_prompt=(
"你是编程学习助手。"
"课程数据为教学模拟数据。"
"需要查询课程时使用工具。"
)
)
这种方式适合大部分工具定义场景。
模型可以看到工具名称、描述和参数 Schema,并决定是否请求调用工具。
方式二:使用普通 Python 函数
LangChain 也可以接受带有适当签名和描述的 Python 函数。
实例
def calculate_sum(a: int, b: int) -> int:
"""计算两个整数之和。"""
return a + b
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[calculate_sum]
)
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "请用工具计算 125 + 375。"
}
]
})
print(result["messages"][-1].content)
这里的函数不需要手动添加 @tool,Agent 创建流程可以将其转换为可执行工具。
方式三:提供商原生工具
create_agent() 也支持某些模型提供商的原生工具定义。
这类工具可能由模型服务端执行,而不是由本地 Python 函数执行。
但其字典结构由具体提供商决定。
注意:不能简单地将任意 MCP 服务器描述字典直接传入 tools,就认为 LangChain 会自动连接 MCP 服务器。
如果需要访问 MCP 工具,应使用兼容的 MCP 集成或适配器,将远程工具转换为 Agent 可用的工具定义,并完成相应的连接、认证和执行配置。
tools=None:不使用工具
实例
agent = create_agent(
model="deepseek:deepseek-flash",
tools=None,
system_prompt="你是一名 Python 编程助手。"
)
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "Python 适合初学者吗?"
}
]
})
print(result["messages"][-1].content)
不传工具时,Agent 不会执行普通客户端工具调用循环。
但它依然是一个编译后的 LangGraph 对象,与直接调用 model.invoke() 的返回结构并不完全相同。
六、system_prompt 参数:系统指令
system_prompt 用于定义 Agent 的角色、任务目标和回答规则。
支持字符串和 SystemMessage 对象。
方式一:字符串
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
system_prompt=(
"你是菜鸟教程 RUNOOB 的编程助手。"
"请使用简洁、准确的中文回答。"
)
)
方式二:SystemMessage
实例
from langchain.agents import create_agent
system_message = SystemMessage(
content=(
"你是一名 Python 编程教程作者。"
"尽量使用初学者能够理解的语言。"
)
)
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
system_prompt=system_message
)
系统提示词通常会在模型调用时添加到消息列表前面。
如果需要根据用户、会话或任务动态生成系统提示词,可以使用 Middleware。
注意:系统提示词不等于可靠的安全边界。身份认证、用户权限和工具执行限制仍需在应用代码中实现。
七、middleware 参数:Agent 中间件
middleware 是 LangChain 1.x Agent 的重要扩展机制。
它可以在 Agent 执行过程中的特定阶段运行自定义逻辑。
例如:
- 在模型调用前修改消息。
- 根据用户身份选择模型。
- 限制模型调用次数。
- 对工具执行进行拦截。
- 处理部分模型或工具错误。
- 记录模型调用和工具执行信息。
官方提供了多种可直接使用的中间件。
实例:限制模型调用次数
实例
from langchain.agents.middleware import (
ModelCallLimitMiddleware
)
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
middleware=[
ModelCallLimitMiddleware(
run_limit=5,
exit_behavior="end"
)
],
system_prompt="你是一名中文编程助手。"
)
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "解释什么是 Python 装饰器。"
}
]
})
print(result["messages"][-1].content)
这里使用 ModelCallLimitMiddleware 限制一次运行中的模型调用次数。
需要注意,模型调用限制不等同于 Token 预算或 API 费用上限。
中间件还可以通过 before_model、after_model、wrap_model_call 和 wrap_tool_call 等机制实现更复杂的行为。
八、response_format 参数:结构化输出
如果希望 Agent 返回特定格式的数据,可以使用 response_format。
例如,让 Agent 返回教程推荐结果,包含推荐课程和理由。
LangChain 支持 Pydantic 模型,以及 ProviderStrategy、ToolStrategy 等结构化输出策略。
实例:使用 Pydantic 定义结构
实例
from langchain.agents import create_agent
class CourseRecommendation(BaseModel):
"""课程推荐结果。"""
course: str = Field(
description="推荐的课程名称"
)
reason: str = Field(
description="推荐理由"
)
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
response_format=CourseRecommendation,
system_prompt=(
"你是一名编程学习顾问。"
"请根据用户的问题返回课程推荐。"
)
)
result = agent.invoke({
"messages": [
{
"role": "user",
"content": (
"我是零基础开发者,"
"想学习 Python,请给出一个推荐。"
)
}
]
})
structured = result["structured_response"]
print("课程:", structured.course)
print("理由:", structured.reason)
如果结构化输出成功,返回状态中会包含:
result["structured_response"]
该字段通常保存根据指定 Schema 解析后的结果。
例如:
{
"course": "Python 基础教程",
"reason": "适合初学者了解 Python 的基本语法。"
}
这里展示的是结果结构示意,并非模型的固定输出。
注意:不同模型对原生结构化输出、Tool Calling 和 JSON Schema 的支持可能不同。
LangChain 可以根据模型能力选择结构化输出策略,但不保证任意模型和任意 Schema 都能成功。
九、state_schema 参数:自定义 Agent 状态
Agent 在执行过程中会保存状态。
最常见的状态字段是:
messages
用于保存用户消息、模型回复和工具执行结果。
AgentState 还可能包含执行控制或结构化响应相关字段,具体以当前 LangChain 版本为准。
如果需要保存额外的应用状态,可以扩展 AgentState。
例如,创建一个学习助手,记录当前用户等级和已完成的主题。
1、定义自定义 State
实例
class LearningAgentState(AgentState):
"""编程学习 Agent 状态。"""
user_level: str
completed_topics: list[str]
其中:
user_level:用户当前学习等级。completed_topics:已完成的学习主题。
但仅仅声明状态字段,并不代表工具执行后这些字段会自动更新。
2、使用 Command 修改 State
如果工具需要修改 Agent 状态,可以返回 LangGraph 的 Command。
不应只对 runtime.state 中的列表执行 append(),就假设状态已经正确提交。
实例
from langchain.messages import ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.types import Command
class LearningAgentState(AgentState):
user_level: str
completed_topics: list[str]
@tool
def track_progress(
topic: str,
runtime: ToolRuntime
) -> Command:
"""将指定主题加入当前已完成列表。"""
completed = list(
runtime.state.get(
"completed_topics", []
)
)
if topic not in completed:
completed.append(topic)
return Command(
update={
"completed_topics": completed,
"messages": [
ToolMessage(
content=(
f"已处理学习主题:{topic}"
),
tool_call_id=runtime.tool_call_id
)
]
}
)
上面使用 Command(update=...) 明确声明需要更新的状态字段。
ToolMessage 通过当前工具调用 ID 与模型的工具请求关联。
3、创建 Agent 并读取更新后的 State
实例
from langchain.agents import create_agent
load_dotenv()
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[track_progress],
state_schema=LearningAgentState,
system_prompt=(
"你是一名学习进度助手。"
"当用户明确要求记录已完成主题时,"
"调用 track_progress 工具。"
)
)
result = agent.invoke({
"messages": [
{
"role": "user",
"content": (
"我已经学完 Python 基础,"
"请帮我记录这个主题。"
)
}
],
"user_level": "入门",
"completed_topics": [
"HTML 基础"
]
})
print("用户等级:", result.get("user_level"))
print("已完成主题:", result.get("completed_topics"))
print("最终回答:")
print(result["messages"][-1].content)
如果模型正确调用 track_progress,更新后的状态可能是:
用户等级: 入门 已完成主题: ['HTML 基础', 'Python 基础']
由于工具是否被调用由模型决定,所以不能认为这个结果一定出现。
如果学习进度必须更新,应在应用层明确执行状态变更,或者在工作流中设置强制执行步骤。
注意:上述方式适用于相对简单的状态更新。如果存在多个工具并行写入同一字段,需要考虑合适的 reducer 或并发控制,避免发生状态更新冲突。
十、context_schema 参数:运行时上下文
context_schema 用于定义 Agent 运行时 Context 的结构。
与 State 不同,Context 通常由应用程序在每次调用时提供,用于保存当前运行需要的可信信息。
例如:
- 当前登录用户 ID。
- 用户所在组织。
- 应用当前使用的语言。
- 业务配置。
实例:传入当前用户身份
实例
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
@dataclass
class UserContext:
user_id: str
@tool
def get_current_user(
runtime: ToolRuntime[UserContext]
) -> str:
"""返回当前请求的模拟用户标识。"""
return runtime.context.user_id
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[get_current_user],
context_schema=UserContext,
system_prompt=(
"你是用户信息助手。"
"查询当前用户标识时使用工具。"
)
)
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "查询我的用户标识。"
}
]
},
context=UserContext(
user_id="user_001"
)
)
print(result["messages"][-1].content)
这里的 user_id 由应用程序提供,而不是模型生成的工具参数。
注意:Context 只是传递运行信息的机制,并不会自动完成用户认证。真实项目应先验证用户登录状态,再构建可信 Context。
十一、checkpointer 参数:会话持久化
如果希望 Agent 能够记住之前的对话,可以配置 checkpointer。
Checkpointer 负责保存和恢复 Agent 的执行状态。
LangGraph 提供适合开发和测试的 InMemorySaver。
实例:保存多轮对话
实例
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
checkpointer=InMemorySaver(),
system_prompt="你是一名编程学习助手。"
)
# 使用相同线程 ID
config = {
"configurable": {
"thread_id": "learning_session_001"
}
}
# 第一轮
agent.invoke(
{
"messages": [
{
"role": "user",
"content": "我正在学习 Python。"
}
]
},
config=config
)
# 第二轮
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "我刚才说我在学习什么?"
}
]
},
config=config
)
print(result["messages"][-1].content)
由于两次请求使用相同的 thread_id,Agent 可以从 Checkpointer 恢复此前的会话消息。
需要注意,InMemorySaver 只保存进程内数据。
程序退出后,内存中的会话状态通常会丢失。
对于需要真正持久化的生产项目,可以使用数据库支持的 Checkpointer。
get_state():查看保存的状态
配置 Checkpointer 后,可以使用 get_state() 获取指定线程的状态快照。
snapshot = agent.get_state(config) print(snapshot.values)
其中,snapshot.values 保存当前已记录的状态数据。
如果没有配置 Checkpointer 或没有对应的线程记录,不能假设 get_state() 一定返回有效的历史状态。
update_state():更新保存的状态
也可以通过:
agent.update_state(
config,
{
"user_level": "进阶"
}
)
修改受状态 Schema 支持的字段。
注意:update_state() 是图状态管理操作,不是单纯修改普通 Python 字典。使用前应确认图已配置适当的检查点存储,并理解状态合并和 reducer 规则。
十二、store 参数:跨会话共享数据
store 用于向 Agent 提供 LangGraph Store。
与 Checkpointer 不同,Store 主要用于跨线程或跨会话共享数据。
例如:
- 用户的长期偏好。
- 学习进度。
- 课程资料。
- 长期记忆。
开发阶段可以使用 InMemoryStore。
实例:创建共享 Store
实例
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# 保存模拟课程数据
store.put(
("demo", "courses"),
"python",
{
"name": "Python 基础",
"level": "入门"
}
)
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
store=store
)
# 直接查看 Store 数据
item = store.get(
("demo", "courses"),
"python"
)
print(item.value)
执行结果:
{'name': 'Python 基础', 'level': '入门'}
当工具需要访问 Store 时,可以结合上一节介绍的 ToolRuntime,使用 runtime.store 读取数据。
注意:InMemoryStore 可以在同一进程内跨调用共享数据,但不是数据库级持久化存储。
十三、interrupt_before 与 interrupt_after
这两个参数用于控制 LangGraph 在指定节点执行前或执行后暂停。
常见节点包括:
model tools
例如,可以在工具执行前暂停,让程序检查工具调用请求。
实例:工具执行前中断
实例
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
@tool
def delete_demo_file(filename: str) -> str:
"""模拟删除文件,不执行实际文件操作。"""
return f"模拟删除文件:{filename}"
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[delete_demo_file],
checkpointer=InMemorySaver(),
interrupt_before=["tools"],
system_prompt=(
"你是文件操作演示助手。"
"用户要求模拟删除文件时使用工具。"
"所有操作均为模拟。"
)
)
config = {
"configurable": {
"thread_id": "demo_delete_001"
}
}
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "请模拟删除 report.txt"
}
]
},
config=config
)
# 查看是否存在待继续执行的节点
snapshot = agent.get_state(config)
print("待执行节点:", snapshot.next)
如果模型发起了工具调用,图可能在 tools 节点执行前暂停。
此时可以检查状态中的工具调用请求。
注意:设置中断并不等于自动完成审批。对于真实删除文件、资金操作等敏感行为,还需要在恢复图执行之前完成可靠的人工确认和权限检查。
如果需要继续执行已经暂停的图,应根据 LangGraph 的恢复机制处理,并使用相同的会话配置。
十四、debug、name、cache 和 transformers
这些参数主要用于调试、集成和扩展执行图。
debug:开启调试输出
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
debug=True
)
启用调试后,LangGraph 可以输出节点执行、状态更新及图流转等信息。
调试输出可能包含用户输入和工具结果,因此生产环境需要注意日志中的敏感数据。
name:指定 Agent 名称
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
name="runoob_learning_agent"
)
名称可以帮助识别执行图,也适用于将 Agent 作为其他 LangGraph 工作流中的子图使用。
cache:图执行缓存
cache 用于配置 LangGraph 的图执行缓存能力。
它不等同于自动缓存所有大语言模型 API 响应。
实际是否发生缓存,还可能取决于节点和缓存策略的配置。
transformers:流事件转换器
transformers 用于注册额外的流事件转换器。
它主要面向需要深度定制流式事件处理流程的开发者。
对于一般的聊天 Agent 或教程示例,通常不需要设置。
十五、create_agent() 的返回值
create_agent() 返回一个 CompiledStateGraph 对象。
这意味着可以使用 LangGraph 提供的同步、异步、流式和状态管理接口。
| 方法 | 作用 |
|---|---|
invoke() |
同步执行 Agent,等待任务结果 |
ainvoke() |
异步执行 Agent |
stream() |
同步流式获取执行结果 |
astream() |
异步流式执行 |
get_state() |
读取已保存的图状态快照 |
update_state() |
更新图状态 |
get_state_history() |
获取受支持的状态历史记录 |
1、invoke():同步调用
实例
"messages": [
{
"role": "user",
"content": "什么是 LangChain?"
}
]
})
print(result["messages"][-1].content)
这是最常用的 Agent 调用方式。
2、ainvoke():异步调用
实例
from langchain.agents import create_agent
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[]
)
async def main():
result = await agent.ainvoke({
"messages": [
{
"role": "user",
"content": "介绍一下 Python。"
}
]
})
print(result["messages"][-1].content)
if __name__ == "__main__":
asyncio.run(main())
适合异步 Web 服务和其他使用 asyncio 的应用。
3、stream():流式输出执行过程
使用 stream_mode="updates" 可以查看 Agent 的节点更新。
实例
from langchain.tools import tool
@tool
def calculate(a: int, b: int) -> int:
"""计算两个整数之和。"""
return a + b
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[calculate]
)
for update in agent.stream(
{
"messages": [
{
"role": "user",
"content": "请使用工具计算 25 + 18。"
}
]
},
stream_mode="updates"
):
print(update)
如果模型调用了工具,流式更新中可能出现模型节点和工具节点的执行结果。
需要注意,stream_mode="updates" 是获取图节点状态更新,而不是逐 Token 输出模型文本。
如果希望实时展示模型生成的 Token,应使用相应的消息流模式。
十六、完整实例:课程学习 Agent
下面将模型、工具、系统提示词和调用流程整合为一个可运行的实例。
这个 Agent 可以查询模拟课程,并根据用户问题返回学习建议。
文件名:create_agent_demo.py
完整实例
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
load_dotenv()
# -------------------------
# 模拟课程数据
# -------------------------
COURSES = {
"python": [
"Python 基础教程",
"Python 数据分析"
],
"java": [
"Java 基础教程",
"Java 面向对象"
],
"html": [
"HTML 基础教程"
]
}
# -------------------------
# 定义工具
# -------------------------
@tool
def search_courses(keyword: str) -> str:
"""按编程语言关键词查询模拟课程。"""
results = COURSES.get(
keyword.lower()
)
if not results:
return "没有找到对应的模拟课程"
return "、".join(results)
@tool
def count_courses(keyword: str) -> str:
"""统计指定语言的模拟课程数量。"""
results = COURSES.get(
keyword.lower(),
[]
)
return (
f"{keyword} 共有 "
f"{len(results)} 门模拟课程"
)
# -------------------------
# 初始化模型
# -------------------------
model = init_chat_model(
"deepseek:deepseek-flash",
timeout=30,
max_retries=2
)
# -------------------------
# 创建 Agent
# -------------------------
agent = create_agent(
model=model,
tools=[
search_courses,
count_courses
],
system_prompt=(
"你是菜鸟教程 RUNOOB 的学习顾问。"
"工具中的数据全部是教学模拟数据。"
"查询课程及数量时优先使用工具。"
"如果工具没有对应数据,"
"不要编造课程名称或数量。"
"请使用简洁、准确的中文回答。"
),
name="runoob_course_agent"
)
# -------------------------
# 执行 Agent
# -------------------------
def ask(question: str):
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": question
}
]
},
config={
"recursion_limit": 15
}
)
print("\n问题:", question)
for message in result["messages"]:
# 模型发起工具调用
if (
message.type == "ai"
and message.tool_calls
):
for call in message.tool_calls:
print(
"[工具调用]",
call["name"],
call["args"]
)
# 工具结果
elif message.type == "tool":
print(
"[工具结果]",
message.name,
message.content
)
print(
"最终回答:",
result["messages"][-1].content
)
if __name__ == "__main__":
ask("有哪些 Python 模拟课程?")
ask("Java 一共有多少门模拟课程?")
ask("学习 HTML 适合从哪里开始?")
运行程序
python create_agent_demo.py
如果 Agent 正确调用对应的工具,可以获取模拟课程数据:
Python 模拟课程: Python 基础教程、Python 数据分析 Java 模拟课程数量: 2 门 HTML 模拟课程: HTML 基础教程
实际的工具调用次数、执行顺序和语言模型回答可能不同。
这里展示的课程数据来自 Python 字典,不会访问菜鸟教程的真实课程数据库。
十七、常见问题
1、create_agent() 和 init_chat_model() 有什么区别?
init_chat_model() 用于创建聊天模型对象。
create_agent() 则在模型基础上增加工具调用、执行循环、状态管理和中间件等能力。
前者主要解决模型初始化,后者主要解决 Agent 执行。
2、可以不传 tools 吗?
可以。
当 tools=None 或为空列表时,Agent 没有可执行的客户端工具调用循环。
但仍可以使用系统提示词和其他支持的执行图功能。
3、create_agent() 是否自动执行工具?
对于注册的可执行客户端工具,Agent 可以根据模型生成的工具调用请求执行它们。
但是否调用工具仍由模型和 Agent 的配置共同决定。
提供商原生工具可能由远程模型服务执行,不能与本地 Python 工具混为一谈。
4、为什么不能直接把 Pydantic 类当作工具使用?
Pydantic 类主要用于定义数据结构和参数 Schema。
它本身不一定提供工具的执行逻辑。
如果希望 Agent 执行 Python 操作,应使用 @tool、可调用函数或其他实现了执行接口的工具对象。
5、为什么 state_schema 中的列表没有更新?
因为读取状态或直接修改状态字典,不等于向图提交了状态更新。
如果工具需要更新自定义 State,应使用受支持的状态更新机制,例如返回 Command。
6、为什么 result["structured_response"] 不存在?
通常是因为没有配置 response_format,或结构化输出过程没有成功完成。
如果需要结构化结果,应正确配置输出 Schema,并确认当前模型支持所需的结构化输出策略。
7、如何让 Agent 记住前面的对话?
可以配置 Checkpointer,并在多次调用时使用相同的 thread_id。
如果需要程序重启后仍能恢复记录,应使用数据库支持的 Checkpointer。
8、store 与 checkpointer 有什么区别?
checkpointer 用于保存和恢复会话状态。
store 用于跨线程或跨会话共享数据,例如用户偏好和长期学习进度。
二者用途不同,不能简单互相代替。
9、interrupt_before=["tools"] 是否自动实现人工审批?
不会。
它只提供工具节点执行前的暂停机制。
实际审批需要由应用程序读取待执行操作、判断权限并决定是否恢复。
10、如何限制 Agent 的执行次数?
可以在运行时通过 recursion_limit 限制图执行步数:
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "你好"
}
]
},
config={
"recursion_limit": 15
}
)
需要注意,recursion_limit 不是严格的工具调用次数限制,也不是 Token 费用上限。
如果需要限制模型调用次数,可以考虑使用 LangChain 中间件。
11、create_agent() 能连接 MCP 吗?
可以通过兼容的 MCP 集成或适配器,将远程 MCP 工具提供给 Agent。
但不能假设任意 MCP 配置字典都可以直接传入 tools 并自动完成远程连接与工具执行。
12、什么时候应该使用 LangGraph 而不是 create_agent()?
如果只是需要模型自动选择工具、处理结果并循环执行,create_agent() 通常已经足够。
如果任务需要明确的节点顺序、复杂分支、多个 Agent 协作或严格的执行控制,可以使用 LangGraph 构建自定义工作流。
create_agent() 本身就是基于 LangGraph 实现的。
总结
create_agent() 是 LangChain 1.x 创建 Agent 的核心入口。
最简单的用法只需要指定模型和工具:
from langchain.agents import create_agent
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[]
)
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "你好"
}
]
})
随着项目复杂度增加,可以逐步配置:
system_prompt:定义 Agent 的角色和行为。middleware:扩展模型调用与工具执行过程。response_format:生成结构化响应。state_schema:定义额外的工作流状态。context_schema:传递可信运行时上下文。checkpointer:保存和恢复会话。store:管理跨会话共享数据。interrupt_before:在指定节点前暂停执行。
注意:create_agent() 负责组织 Agent 的执行流程,模型负责提出工具调用请求,工具负责真正执行操作,而 State 与 Middleware 则帮助管理整个执行过程。
