本文深入解析大模型 Agent 的核心概念与开发流程，重点介绍基于 LangChain 框架实现工具调用与自主决策的方法。内容涵盖 ReAct 模式原理、Prompt 工程优化、自定义工具构建及完整代码示例，旨在帮助开发者快速掌握 Agent 应用开发技能，解决复杂任务规划问题。文章详细阐述了从工具定义、调度原理到标准 Agent 初始化的全过程，并补充了错误处理、Token 限制及安全注意事项，提供了一套完整的实战指南。
大模型 Agent（智能体）是结合了大规模神经网络模型和自主计算实体的技术，它具备强大的表达、学习和交互能力，能够在无人干预的情况下，根据环境信息自主决策和控制行为。简单来说，Agent 是增强大模型能力的技术方案路径。
一个完整的 Agent 系统通常包含以下三个核心部分：
Agent 的运行大体遵循以下闭环流程：
这一过程会循环执行，直到大模型认为已经获得足够的信息来回答用户问题。
LangChain 是目前最流行的 LLM 应用开发框架之一，它提供了丰富的组件来构建 Agent。在 LangChain 中，我们可以灵活地定义工具、配置提示词模板以及管理对话状态。
在 LangChain 中，我们需要为每个工具定义清晰的描述和参数。这有助于大模型理解何时以及如何调用该工具。
以下是一个基于天气查询的自定义工具示例，展示了如何使用 Pydantic 进行参数校验：
from pydantic import BaseModel, Field
import requests
# 模拟 API Key 配置
SENIVERSE_API_KEY = "your_api_key_here"
def weather(location: str, api_key: str):
url = f"https://api.seniverse.com/v3/weather/now.json?key={api_key}&location={location}&language=zh-Hans&unit=c"
try:
response = requests.get(url)
if response.status_code == 200:
data = response.json()
weather_info = {
"temperature": data["results"][0]["now"]["temperature"],
"description": data["results"][0]["now"]["text"],
}
return weather_info
else:
raise Exception(f"Failed to retrieve weather: {response.status_code}")
except Exception as e:
return f"Error: {str(e)}"
def weathercheck(location: str):
return weather(location, SENIVERSE_API_KEY)
class WeatherInput(BaseModel):
location: str = Field(description="City name, include city and county")
为了让大模型能够正确调用上述工具，我们需要构建一套 Prompt 机制，告诉模型可用的工具列表及其描述。这里采用 ReAct（Reason + Act）模式。
ReAct 是由单词'Reason'和'Act'组合而成。前者对应于推理，即大模型的通用文本逻辑判断能力；后者对应于行动，即具备专业知识的特定领域精确回答能力，或者说是调用外部工具的能力。ReAct 把思考和行动相结合，通过二者的依次迭代执行完成任务。
我们需要将工具元数据转换为模型可理解的格式：
from langchain.utilities import ArxivAPIWrapper
from langchain_experimental.tools import PythonAstREPLTool
from typing import Dict, Tuple
import os
import json
arxiv = ArxivAPIWrapper()
python_tool = PythonAstREPLTool()
class ElectricityBillTool:
def run(self, name, start_date, end_date):
# 实际场景中应连接数据库查询电费
return f"电费查询结果：姓名：{name}, 期间：{start_date} 到 {end_date}, 电费：100 元"
def tool_wrapper_for_model(tool, expects_kwargs=True):
def tool_(args_json):
args = json.loads(args_json)
if expects_kwargs:
return tool.run(**args)
else:
return tool.run(args['query'])
return tool_
TOOLS = [
{
'name_for_human': 'Arxiv',
'name_for_model': 'Arxiv',
'description_for_model': 'A wrapper around Arxiv.org Useful for when you need to answer questions about Physics, Mathematics...',
'parameters': [{"name": "query", "type": "string", "description": "the document id of arxiv to search", : }],
: tool_wrapper_for_model(arxiv, expects_kwargs=)
},
{
: ,
: ,
: ,
: [
{: , : , : , : },
{: , : , : , : },
{: , : , : , : }
],
: tool_wrapper_for_model(ElectricityBillTool(), expects_kwargs=)
}
]
TOOL_DESC =
REACT_PROMPT =
下面展示了一个完整的 Agent 主循环逻辑，包括 Prompt 构建、模型响应获取、工具解析与执行。
def build_planning_prompt(TOOLS, query):
tool_descs = []
tool_names = []
for info in TOOLS:
tool_descs.append(
TOOL_DESC.format(
name_for_model=info['name_for_model'],
name_for_human=info['name_for_human'],
description_for_model=info['description_for_model'],
parameters=json.dumps(info['parameters'], ensure_ascii=False),
)
)
tool_names.append(info['name_for_model'])
tool_descs = '\n\n'.join(tool_descs)
tool_names = ','.join(tool_names)
prompt = REACT_PROMPT.format(tool_descs=tool_descs, tool_names=tool_names, query=query)
return prompt
def parse_latest_plugin_call(text: str) -> Tuple[str, str]:
i = text.rfind('\nAction:')
j = text.rfind('\nAction Input:')
k = text.rfind('\nObservation:')
if 0 <= i < j:
if k < j:
text = text.rstrip() + '\nObservation:'
k = text.rfind('\nObservation:')
if 0 <= i < j < k:
plugin_name = text[i + len('\nAction:'):j].strip()
plugin_args = text[j + len('\nAction Input:'):k].strip()
return plugin_name, plugin_args
return '', ''
def ():
use_toolname, action_input = parse_latest_plugin_call(response)
use_toolname == :
used_tool_meta = (( x: x[] == use_toolname, tools))
(used_tool_meta) == :
api_output = used_tool_meta[][](action_input)
api_output
():
openai OpenAI
client = OpenAI()
completion = client.chat.completions.create(
model=,
messages=[{: , : prompt}],
stream=,
stop=stop,
)
response = completion.choices[].message.content
response
():
prompt = build_planning_prompt(choose_tools, query)
stop = [, ]
()
(prompt)
response = get_model_response(prompt, stop)
response:
api_output = use_api(choose_tools, response)
api_output = (api_output)
== api_output:
()
()
prompt = prompt + response + + api_output
response = get_model_response(prompt, stop)
( + response + )
__name__ == :
query =
choose_tools = TOOLS
( * )
main(query, choose_tools)
除了手动构建 Prompt 循环，LangChain 也提供了封装好的 Agent 类型，简化开发流程。
使用 initialize_agent 函数可以快速创建 Agent，指定使用的工具链和模型。
from langchain.agents import initialize_agent, AgentType
from langchain_community.llms import ChatOpenAI
from langchain.tools import Tool
def get_llm():
api_key = os.getenv("PROXY_API_KEY")
api_url = os.getenv("PROXY_SERVER_URL")
api_url = api_url.split('/chat')[0]
model = os.getenv("PROXYLLM_BACKEND")
if "openai" in api_url:
llm = ChatOpenAI(temperature=0, model=model, openai_api_base=api_url, openai_api_key=api_key)
elif "bigmodel" in api_url:
from langchain_community.chat_models import ChatZhipuAI
llm = ChatZhipuAI(temperature=0.01, api_key=api_key, model="glm-4")
elif "dashscope" in api_url:
from langchain_community.chat_models.tongyi import ChatTongyi
llm = ChatTongyi(model="qwen-max", top_p=0.01, streaming=True, dashscope_api_key=api_key)
return llm
def get_tools():
llm = get_llm()
tools = [
Tool(
name="查询负荷数据及同比环比",
func=calculate_growth,
description="当你想要查询某地负荷（最大值、最小值、平均值），查询最大/最小值发生时间，或者查询同比或环比变化的时候很有用。输入格式：地区 | 时间 | 指标 | 对比类型。例如：上海|2023|最大 | 环比。"
),
Tool(
name="查询负荷新高 (低)",
func=find_record_breaking_loads,
description="当你想要查询某地负荷几创新高或几创新低的时候很有用。输入格式：地区 | 基础时间 | 对比时间 | 高/低。"
),
]
tools
():
datetime datetime
current_time = datetime.now()
prompt = (
+ query
)
prompt
():
llm = get_llm()
tools = get_tools()
agent = initialize_agent(
tools,
llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=,
max_iterations=,
handle_parsing_errors=,
)
agent
agent = get_tools_agent()
prompt = get_tools_agent_prompt()
response = agent.run(prompt)
(response)
在实际业务场景中，工具函数的实现需要处理数据解析和业务逻辑。
import pandas as pd
from dateutil.parser import parse
df = pd.DataFrame({'公司': ['北京', '上海'], '时间': ['2023-01-01', '2023-01-02'], '负荷': [100, 120]})
def calculate_growth(context):
params = context.split("|")
if len(params) < 3:
return "至少需要三个参数（region, time_period, compare_type）。"
region = params[0]
time_period = params[1]
compare_type = params[2]
growth_type = params[3] if len(params) > 3 else None
# 此处应接入真实数据库查询逻辑
return f"{region} 在{time_period}的{compare_type}增长率为 5%。"
def find_record_breaking_loads(context):
params = context.split("|")
if len(params) not in [3, 4]:
return "您输入的参数数量不匹配，请仔细核对。"
region = params[0]
current_time_period = params[1]
high_low = params[-1]
compare_time_period = params[] (params) == ((current_time_period[:])-) + current_time_period[:]
df_region = df[df[] == region]
Agent 在执行过程中可能会遇到工具调用失败、API 超时等情况。务必在工具函数内部增加 try-except 块，并返回友好的错误信息，以便大模型能根据错误信息调整后续策略。
长轮询会导致 Token 消耗迅速增加。建议设置 max_iterations 限制最大思考步数，避免死循环。同时注意控制 Prompt 的长度，必要时对历史对话进行摘要压缩。
不要将敏感信息（如 API Key、数据库密码）硬编码在代码中。应使用环境变量管理密钥。此外，对用户输入进行校验，防止注入攻击。
本文详细介绍了大模型 Agent 的核心概念、基于 LangChain 的开发流程以及代码实现细节。通过掌握 ReAct 模式和工具调用机制，开发者可以构建出具备自主决策能力的智能应用。未来随着多模态能力和更长上下文的普及，Agent 的应用场景将更加广泛。建议开发者在实践中不断迭代 Prompt 工程，优化工具描述，以提升 Agent 的准确率和稳定性。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online