大模型 Agent 开发指南：基于 LangChain 的实战应用

2.2 工具调度原理

为了让大模型能够正确调用上述工具，我们需要构建一套 Prompt 机制，告诉模型可用的工具列表及其描述。这里采用 ReAct（Reason + Act）模式。

2.2.1 ReAct 模式详解

ReAct 是由单词'Reason'和'Act'组合而成。前者对应于推理，即大模型的通用文本逻辑判断能力；后者对应于行动，即具备专业知识的特定领域精确回答能力，或者说是调用外部工具的能力。ReAct 把思考和行动相结合，通过二者的依次迭代执行完成任务。

2.2.2 工具描述构建

我们需要将工具元数据转换为模型可理解的格式：

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", 'required': True}], 
        'tool_api': tool_wrapper_for_model(arxiv, expects_kwargs=False)
    },
    {
        'name_for_human': 'ElectricityBill',
        'name_for_model': 'ElectricityBill',
        'description_for_model': '查询电费工具，根据姓名、开始时间和结束时间查询电费。',
        'parameters': [
            {"name": "name", "type": "string", "description": "姓名", 'required': True},
            {"name": "start_date", "type": "string", "description": "开始时间", 'required': True},
            {"name": "end_date", "type": "string", "description": "结束时间", 'required': True}
        ],
        'tool_api': tool_wrapper_for_model(ElectricityBillTool(), expects_kwargs=True)
    }
]

TOOL_DESC = "{name_for_model}: Call this tool to interact with the {name_for_human} API. What is the {name_for_human} API useful for? {description_for_model} Parameters: {parameters} Format the arguments as a JSON object."

REACT_PROMPT = """Answer the following questions as best you can. You have access to the following tools:

{tool_descs}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {query}"""

2.3 完整 Agent 实现流程

下面展示了一个完整的 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_api(tools, response):
    use_toolname, action_input = parse_latest_plugin_call(response)
    if use_toolname == "":
        return "no tool founds"
    used_tool_meta = list(filter(lambda x: x["name_for_model"] == use_toolname, tools))
    if len(used_tool_meta) == 0:
        return "no tool founds"
    api_output = used_tool_meta[0]["tool_api"](action_input)
    return api_output

def get_model_response(prompt, stop):
    from openai import OpenAI
    client = OpenAI()
    completion = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=False,
        stop=stop,
    )
    response = completion.choices[0].message.content
    return response

def main(query, choose_tools):
    prompt = build_planning_prompt(choose_tools, query)
    stop = ["Observation:", "Observation:\n"]
    print("Initial Prompt:")
    print(prompt)
    response = get_model_response(prompt, stop)

    while "Final Answer:" not in response:
        api_output = use_api(choose_tools, response)
        api_output = str(api_output)
        if "no tool founds" == api_output:
            break
        print(f"Response: \033[32m{response}\033[0m")
        print(f"Output: \033[34m{api_output}\033[0m")
        prompt = prompt + response + ' ' + api_output
        response = get_model_response(prompt, stop)

    print("\033[32m" + response + "\033[0m")

if __name__ == "__main__":
    query = "查一下张三 2024 年 1 月的电费"
    choose_tools = TOOLS
    print("=" * 10)
    main(query, choose_tools)

三、LangChain 标准 Agent 开发模式

除了手动构建 Prompt 循环，LangChain 也提供了封装好的 Agent 类型，简化开发流程。

3.1 初始化 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="当你想要查询某地负荷几创新高或几创新低的时候很有用。输入格式：地区 | 基础时间 | 对比时间 | 高/低。"
        ),
    ]
    return tools

def get_tools_agent_prompt(query, history=None):
    from datetime import datetime
    current_time = datetime.now()
    prompt = (
        f"""
        如果查询未指定年/月，请参考当前时刻：{current_time}。
        请用中文思考及回复。
        优先考虑使用工具解决问题，当工具无法解决时，尝试根据经验直接回答。
        最终回复时若内容较多需要注意保持可读性，添加合理的换行。
        下面是用户问题:
        """
        + query
    )
    return prompt

def get_tools_agent():
    llm = get_llm()
    tools = get_tools()
    agent = initialize_agent(
        tools, 
        llm, 
        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, 
        verbose=True, 
        max_iterations=5,
        handle_parsing_errors=True,
    )
    return agent

# 使用示例
agent = get_tools_agent()
prompt = get_tools_agent_prompt("查询北京 2023 年夏季最高负荷")
response = agent.run(prompt)
print(response)

3.2 关键 Agent 类型说明

• Zero-shot ReAct Description：零样本，无记忆。在运行时，只考虑与当前代理的一次交互，不保留对话历史。适用于单次任务完成。
• Conversational：引入了对话历史，因为有记忆了，所以需要初始化代理时引入 memory 参数。其一个缺点是可能无法执行复杂的 Tool 调用任务。
• Chat：常规情况下以 OpenAI 方式初始化 LLM，此类 Agent 可以以 ChatOpenAI 方式初始化模型。更专注于对话式交互。

四、工具函数实现细节

在实际业务场景中，工具函数的实现需要处理数据解析和业务逻辑。

4.1 负荷数据查询工具

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[2] if len(params) == 4 else str(int(current_time_period[:4])-1) + current_time_period[4:]
    
    df_region = df[df['公司'] == region]
    # 模拟计算逻辑
    return f"{region}在{current_time_period}相较{compare_time_period}共有 1 次创下新{high_low}。"

五、最佳实践与注意事项

5.1 错误处理

Agent 在执行过程中可能会遇到工具调用失败、API 超时等情况。务必在工具函数内部增加 try-except 块，并返回友好的错误信息，以便大模型能根据错误信息调整后续策略。

5.2 Token 限制

长轮询会导致 Token 消耗迅速增加。建议设置 max_iterations 限制最大思考步数，避免死循环。同时注意控制 Prompt 的长度，必要时对历史对话进行摘要压缩。

5.3 安全性

不要将敏感信息（如 API Key、数据库密码）硬编码在代码中。应使用环境变量管理密钥。此外，对用户输入进行校验，防止注入攻击。

六、总结

本文详细介绍了大模型 Agent 的核心概念、基于 LangChain 的开发流程以及代码实现细节。通过掌握 ReAct 模式和工具调用机制，开发者可以构建出具备自主决策能力的智能应用。未来随着多模态能力和更长上下文的普及，Agent 的应用场景将更加广泛。建议开发者在实践中不断迭代 Prompt 工程，优化工具描述，以提升 Agent 的准确率和稳定性。