Agent模式学习笔记

1

LLM问答 → 单Agent → 多Agent协作

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

┌─────────────────────────────────┐
│         Agent 核心架构           │
├─────────────────────────────────┤
│  Planning (规划层)               │
│  ├─ 任务分解                     │
│  └─ 策略制定                     │
├─────────────────────────────────┤
│  Memory (记忆层)                 │
│  ├─ 短期记忆 (对话上下文)        │
│  └─ 长期记忆 (知识库)            │
├─────────────────────────────────┤
│  Tools (工具层)                  │
│  ├─ 搜索工具                     │
│  ├─ 计算工具                     │
│  └─ API调用                      │
├─────────────────────────────────┤
│  Action (执行层)                 │
│  └─ 工具调用与结果处理           │
└─────────────────────────────────┘

1
2

用户：北京今天天气如何？
LLM：我无法提供实时信息，建议你通过天气网站查询...

1
2
3
4

用户：北京今天天气如何？
LLM：[调用 get_current_weather(location="北京")]
工具返回：北京今天是雨天
LLM：今天北京是雨天，请记得带伞！

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 简化示例
system_prompt = f"""
你是智能助手。当需要查询天气时，请使用以下格式：
<tool_call>
{{"name": "get_current_weather", "arguments": {{"location": "城市名"}}}}
</tool_call>

可用工具：
{json.dumps(tools)}
"""

1
2
3
4

# 用户问题 → 调用1个工具 → 返回结果
用户："北京天气怎么样？"
→ get_current_weather(location="北京")
→ "北京今天是晴天"

1
2
3
4

# 从多个工具中选择合适的
工具列表：[天气查询, 数学计算, 翻译, 搜索]
用户："帮我算一下 15452 * 88"
→ LLM选择 calculate_math(expression="15452 * 88")

1
2
3
4

# 工具调用形成链条，前一个结果影响后一个
用户："查询阿里巴巴财报并发邮件给老板"
→ Step1: web_search(query="阿里巴巴2025财报")
→ Step2: send_email(recipient="boss@...", body="财报摘要...")

1
2
3
4
5
6
7

# 同时调用多个工具，提高效率
用户："查询北京和巴黎的天气和新闻"
→ 并行执行：
   - get_weather(location="北京")
   - get_weather(location="巴黎")
   - get_news(location="北京")
   - get_news(location="巴黎")

1
2

Step 1 → Step 2 → Step 3 → Step 4
(分析)   (大纲)   (写作)   (优化)

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

# 伪代码示例
def blog_workflow(topic):
    # Step 1: 主题分析
    analysis = llm_call("分析主题", topic)

    # Step 2: 生成大纲
    outline = llm_call("生成大纲", analysis)

    # Step 3: 撰写内容
    draft = llm_call("撰写文章", outline)

    # Step 4: SEO优化
    final = llm_call("优化发布", draft)

    return final

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

┌─────────────────────────────────┐
│  1. Planning (规划下一步)        │
│     ↓                            │
│  2. Acting (执行工具)            │
│     ↓                            │
│  3. Reflecting (反思结果)        │
│     ↓                            │
│  4. 判断是否完成？               │
│     ├─ 是 → 输出最终答案         │
│     └─ 否 → 回到步骤1            │
└─────────────────────────────────┘

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

class ReActAgent:
    def run(self, user_query):
        while not task_completed:
            # 1. Planning
            plan = planner_llm(history, user_query)

            # 2. Acting
            result = execute_tool(plan.tool_name, plan.args)

            # 3. Reflecting
            reflection = reflector_llm(plan, result)

            # 4. 更新历史
            history.append({
                "thought": plan.thought,
                "action": plan.tool_name,
                "observation": result,
                "reflection": reflection
            })

            # 5. 判断是否完成
            if plan.decision == "finish":
                return plan.final_answer

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

        ┌─────────────┐
        │ 中央调度器   │
        └──────┬──────┘
               │
    ┌──────────┼──────────┐
    ↓          ↓          ↓
┌───────┐ ┌───────┐ ┌───────┐
│Agent1 │ │Agent2 │ │Agent3 │
│(搜索) │ │(分析) │ │(写作) │
└───────┘ └───────┘ └───────┘

1
2
3
4
5
6
7
8

┌──────────────┐      ┌──────────────┐
│   Redis      │      │   MySQL      │
│  (工作区)    │ ←──→ │  (持久化)    │
├──────────────┤      ├──────────────┤
│ 活跃会话     │      │ 历史记录     │
│ 快速读写     │      │ 长期保存     │
│ 自动过期     │      │ 可查询       │
└──────────────┘      └──────────────┘

1
2
3
4
5
6

messages = [
    {"role": "system", "content": "你是助手..."},
    {"role": "user", "content": "用户问题"},
    {"role": "assistant", "content": "AI回复"},
    {"role": "tool", "tool_call_id": "xxx", "content": "工具结果"}
]

 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

# 工具定义
def get_weather(location):
    return f"{location}今天是晴天，22℃"

# 工具Schema
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询城市天气",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string"}
            },
            "required": ["location"]
        }
    }
}]

# 调用流程
messages = [{"role": "user", "content": "北京天气？"}]
response = client.chat.completions.create(
    model="qwen-plus",
    messages=messages,
    tools=tools
)

# 处理工具调用
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    result = get_weather(**json.loads(tool_call.function.arguments))
    # 将结果返回给LLM...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 任务：搜索信息 → 生成报告 → 发送邮件
def chain_agent(query):
    # Step 1: 搜索
    search_result = web_search(query)

    # Step 2: 生成报告（LLM自动总结）
    # LLM会将search_result作为上下文，生成报告内容

    # Step 3: 发送邮件
    # LLM会调用send_email工具，参数中包含生成的报告