1
2
3

传统 LLM：用户 → LLM → 回答

Agent：用户 → LLM → 思考 → 调用工具 → 观察结果 → 再思考 → ... → 最终回答

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

┌─────────────────────────────────────────────────┐
│                    Agent                         │
│  ┌─────────────┐    ┌──────────────────────┐   │
│  │    LLM      │◄──►│   AgentExecutor      │   │
│  │  (大脑)     │    │   (执行器)            │   │
│  └─────────────┘    └──────────┬───────────┘   │
│                                │                │
│                     ┌──────────▼───────────┐   │
│                     │       Tools          │   │
│                     │  ┌────┐ ┌────┐ ...  │   │
│                     │  │搜索│ │计算│      │   │
│                     │  └────┘ └────┘      │   │
│                     └──────────────────────┘   │
└─────────────────────────────────────────────────┘

1

pip install langchain langchain-openai python-dotenv

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from langchain.agents import initialize_agent, load_tools
from langchain_openai import ChatOpenAI
import os

# 设置 API Key
os.environ["OPENAI_API_KEY"] = "your-api-key"

# 1. 初始化 LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 2. 加载工具
tools = load_tools(["llm-math", "serpapi"], llm=llm)

# 3. 初始化 Agent
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent="zero-shot-react-description",
    verbose=True  # 打印思考过程
)

# 4. 运行
result = agent.run("珠穆朗玛峰的高度是多少米？换算成英尺是多少？")
print(result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

> Entering new AgentExecutor chain...
我需要先找到珠穆朗玛峰的高度，然后换算成英尺
Action: Search
Action Input: 珠穆朗玛峰高度
Observation: 8848.86米
Thought: 现在我需要把8848.86米换算成英尺
Action: Calculator
Action Input: 8848.86 * 3.28084
Observation: Answer: 29031.8
Thought: 我现在知道最终答案了
Final Answer: 珠穆朗玛峰高度为8848.86米，约合29031.8英尺

> Finished chain.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的天气信息
    
    Args:
        city: 城市名称，如"北京"、"上海"
    
    Returns:
        天气信息字符串
    """
    # 这里模拟天气查询，实际可调用天气 API
    weather_data = {
        "北京": "晴天，温度 15°C，空气质量良好",
        "上海": "多云，温度 18°C，有轻微雾霾",
        "深圳": "小雨，温度 22°C，湿度较高"
    }
    return weather_data.get(city, f"未找到{city}的天气信息")

@tool
def calculate_bmi(height: float, weight: float) -> str:
    """
    计算 BMI 指数
    
    Args:
        height: 身高（米）
        weight: 体重（公斤）
    
    Returns:
        BMI 值和健康建议
    """
    bmi = weight / (height ** 2)
    
    if bmi < 18.5:
        status = "偏瘦"
        advice = "建议增加营养摄入"
    elif bmi < 24:
        status = "正常"
        advice = "保持良好的生活习惯"
    elif bmi < 28:
        status = "超重"
        advice = "建议适当运动，控制饮食"
    else:
        status = "肥胖"
        advice = "建议咨询医生，制定减重计划"
    
    return f"BMI: {bmi:.1f}，状态：{status}，{advice}"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from langchain.agents import initialize_agent
from langchain_openai import ChatOpenAI

# 初始化
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 使用自定义工具
tools = [get_weather, calculate_bmi]

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent="zero-shot-react-description",
    verbose=True
)

# 测试
result = agent.run("我身高1.75米，体重70公斤，我的BMI是多少？北京今天天气怎么样？")
print(result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

Question: 用户问题
  ↓
Thought: 思考下一步做什么
  ↓
Action: 选择工具并执行
  ↓
Observation: 观察执行结果
  ↓
Thought: 根据结果继续思考
  ↓
... (循环直到得出答案)
  ↓
Final Answer: 最终答案

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# AgentExecutor 的核心逻辑（简化版）
def agent_loop(agent, tools, input):
    while True:
        # 1. Agent 决定下一步行动
        action = agent.plan(input)
        
        # 2. 如果是最终答案，返回结果
        if action.is_final_answer():
            return action.answer
        
        # 3. 否则执行工具
        tool = find_tool(action.tool_name, tools)
        observation = tool.run(action.tool_input)
        
        # 4. 更新输入，继续循环
        input = update_input(input, action, observation)

1
2
3
4
5
6
7
8
9

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent="zero-shot-react-description",
    max_iterations=10,      # 最大迭代次数
    max_execution_time=30,  # 最大执行时间（秒）
    early_stopping_method="generate",  # 超时处理方式
    verbose=True
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from langchain.memory import ConversationBufferMemory
from langchain.agents import AgentExecutor

# 创建记忆组件
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 创建带记忆的 Agent
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent="conversational-react-description",
    memory=memory,
    verbose=True
)

# 多轮对话
agent.run("我叫张三")
agent.run("我刚才告诉你我叫什么？")  # 能记住之前的对话

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

import os
from typing import Optional
from langchain.agents import initialize_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.memory import ConversationBufferMemory

# ========== 工具定义 ==========

@tool
def search_web(query: str) -> str:
    """搜索网络信息，返回相关结果"""
    # 实际可接入 SerpAPI 或 Google Search API
    return f"搜索 '{query}' 的结果：[模拟搜索结果...]"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """
    发送邮件
    
    Args:
        to: 收件人邮箱
        subject: 邮件主题
        body: 邮件正文
    """
    # 实际可接入 SMTP 或邮件 API
    return f"邮件已发送给 {to}，主题：{subject}"

@tool
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    """获取当前时间"""
    from datetime import datetime
    import pytz
    tz = pytz.timezone(timezone)
    return datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")

@tool
def calculate(expression: str) -> str:
    """
    计算数学表达式
    
    Args:
        expression: 数学表达式，如 "2+2", "100*5"
    """
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# ========== Agent 配置 ==========

def create_smart_assistant():
    # 初始化 LLM
    llm = ChatOpenAI(
        model="gpt-4",
        temperature=0,
        openai_api_key=os.getenv("OPENAI_API_KEY")
    )
    
    # 工具列表
    tools = [search_web, send_email, get_current_time, calculate, get_weather]
    
    # 记忆
    memory = ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True
    )
    
    # 创建 Agent
    agent = initialize_agent(
        tools=tools,
        llm=llm,
        agent="conversational-react-description",
        memory=memory,
        verbose=True,
        max_iterations=10,
        handle_parsing_errors=True
    )
    
    return agent

# ========== 使用示例 ==========

if __name__ == "__main__":
    assistant = create_smart_assistant()
    
    # 场景1：查询天气
    print(assistant.run("北京今天天气怎么样？"))
    
    # 场景2：复杂任务
    print(assistant.run("""
    请帮我完成以下任务：
    1. 查询当前时间
    2. 查询上海天气
    3. 如果上海下雨，发邮件给 test@example.com 提醒带伞
    """))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# ✅ 好的 Tool 设计
@tool
def search_database(query: str) -> str:
    """
    在产品数据库中搜索信息
    
    Args:
        query: 搜索关键词
    
    Returns:
        匹配的产品信息列表
    """
    pass

# ❌ 不好的设计：描述不清晰
@tool  
def search(q: str) -> str:
    """搜索"""
    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

from langchain.agents import load_tools, initialize_agent
from langchain_openai import ChatOpenAI

# 自定义前缀 Prompt
prefix = """
你是一个智能助手，可以帮助用户完成各种任务。
请根据用户的问题，选择合适的工具来解决问题。

可用工具：
"""

# 自定义后缀 Prompt  
suffix = """
开始！

用户问题: {input}
{agent_scratchpad}
"""

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent="zero-shot-react-description",
    agent_kwargs={
        "prefix": prefix,
        "suffix": suffix
    }
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,  # 自动处理解析错误
    max_iterations=5,            # 限制迭代次数
    early_stopping_method="generate"  # 超时生成答案
)

# 带异常处理的调用
try:
    result = agent_executor.invoke({"input": "你的问题"})
except Exception as e:
    print(f"执行出错: {e}")
    result = "抱歉，处理您的请求时出现了问题。"

1
2
3
4

import langchain
langchain.debug = True  # 开启详细日志

agent.run("你的问题")

1
2
3
4
5
6
7
8
9

import os

# 配置 LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"

# Agent 执行会自动记录到 LangSmith
agent.run("你的问题")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

from langchain.callbacks.base import BaseCallbackHandler

class MyCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"LLM 开始调用，提示词：{prompts}")
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"工具开始调用：{serialized['name']}")
    
    def on_tool_end(self, output, **kwargs):
        print(f"工具返回结果：{output}")

agent = initialize_agent(
    tools=tools,
    llm=llm,
    callbacks=[MyCallbackHandler()],
    verbose=True
)