智能体架构设计：从简单到复杂

 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

# file: level1_single_call.py
# 单次LLM调用示例：中英互译
import os
from openai import OpenAI

# 初始化客户端，API Key从环境变量读取（避免硬编码泄露密钥）
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

def translate(text: str, direction: str = "zh2en") -> str:
    """根据方向完成中英互译"""
    # 根据方向决定源语言和目标语言
    src, dst = ("中文", "英文") if direction == "zh2en" else ("英文", "中文")
    # 构造系统提示，限定LLM的角色与输出格式
    system_prompt = f"你是一个专业翻译，负责将{src}翻译成{dst}，只输出译文。"

    # 单次调用：一次system + 一次user，得到一次response
    response = client.chat.completions.create(
        model="gpt-5.4",  # 模型名可按你账号可用的模型替换
        messages=[
            {"role": "system", "content": system_prompt},  # 设定角色
            {"role": "user", "content": text},             # 用户输入
        ],
        temperature=0.3,  # 低温度让翻译更稳定、少发挥
    )
    # choices[0].message.content 即模型生成的文本
    return response.choices[0].message.content.strip()

if __name__ == "__main__":
    print(translate("人工智能正在改变软件开发的范式。", "zh2en"))
    # 输出示例：Artificial intelligence is changing the paradigm of software development.

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

用户输入
   │
   ▼
┌─────────────────────────────┐
│  LLM 推理：Thought → Action │  ← 循环开始
└─────────────────────────────┘
   │ Action（如 search("天气")）
   ▼
┌─────────────────────────────┐
│      工具执行 / 环境反馈       │
└─────────────────────────────┘
   │ Observation（如 "北京 25℃ 晴"）
   ▼
┌─────────────────────────────┐
│  LLM 再推理：是否已能回答？    │
└─────────────────────────────┘
   │
   ├── 否 → 回到循环顶部
   └── 是 → 输出 Final Answer

  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
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

# file: level2_react_agent.py
# ReAct架构智能体：纯Python实现，不依赖LangChain
import os
import re
from openai import OpenAI

# 初始化客户端，密钥从环境变量读取
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# ---------- 工具定义 ----------
def calculator(expression: str) -> str:
    """简易计算器，仅支持四则运算"""
    try:
        # 用正则限制字符集，防止恶意注入（如导入os）
        if not re.match(r"^[\d\s\+\-\*/\.\(\)]+$", expression):
            return "错误：表达式包含非法字符"
        # 字符集已限定，此处eval相对安全
        return str(eval(expression))
    except Exception as e:
        return f"计算出错：{e}"

def search(query: str) -> str:
    """模拟搜索工具（真实场景可接入搜索API）"""
    # 这里用一个字典假装是搜索引擎的返回
    fake_db = {
        "北京天气": "北京今日 25℃，晴，微风",
        "上海天气": "上海今日 28℃，多云",
        "Python版本": "Python最新稳定版为 3.13",
    }
    return fake_db.get(query.strip(), f"未找到关于'{query}'的信息")

# 工具注册表：名称 -> (函数, 描述)
# 描述会拼进提示词，让LLM知道有哪些工具可用
TOOLS = {
    "calculator": (calculator, "四则运算计算器，输入数学表达式，如 3*8+2"),
    "search": (search, "信息搜索工具，输入关键词查询天气、常识等"),
}

# ---------- ReAct 提示模板 ----------
REACT_PROMPT = """你是一个使用ReAct模式工作的智能体。
可用工具：
{tool_descriptions}

请严格按以下格式输出，每一步用关键词分隔：

Thought: 你的思考过程
Action: 工具名称
Action Input: 工具输入参数（字符串）

当你已得到足够信息可以回答用户问题时，输出：

Thought: 我已经可以回答了
Final Answer: 最终答案

问题：{question}
"""

def build_tool_description() -> str:
    """构造工具说明文本，拼成提示词的一部分"""
    lines = []
    for name, (_, desc) in TOOLS.items():
        lines.append(f"- {name}: {desc}")
    return "\n".join(lines)

def parse_action(text: str):
    """从LLM输出中解析出 Action 和 Action Input，或 Final Answer"""
    # 用正则匹配各关键字后的内容
    action_match = re.search(r"Action:\s*(.+)", text)
    input_match = re.search(r"Action Input:\s*(.+)", text)
    final_match = re.search(r"Final Answer:\s*([\s\S]+)", text)
    # 优先判断是否已给出最终答案
    if final_match:
        return "final", final_match.group(1).strip()
    if action_match and input_match:
        return action_match.group(1).strip(), input_match.group(1).strip()
    return None, None  # 格式不符时返回None，由主循环兜底处理

def react_agent(question: str, max_steps: int = 5) -> str:
    """ReAct主循环：思考-行动-观察，直到得出答案或步数耗尽"""
    messages = [
        {"role": "system", "content": REACT_PROMPT.format(
            tool_descriptions=build_tool_description(),  # 注入工具清单
            question=question,                            # 注入用户问题
        )},
    ]

    for step in range(max_steps):
        # 1. LLM推理：生成 Thought + Action
        resp = client.chat.completions.create(
            model="gpt-5.4",
            messages=messages,
            temperature=0.2,  # 低温度让格式更稳定
        )
        output = resp.choices[0].message.content
        print(f"\n===== Step {step + 1} =====")
        print(output)

        # 2. 解析动作（要么是工具调用，要么是final）
        action, action_input = parse_action(output)

        # 3. 若给出最终答案，结束循环
        if action == "final":
            return action_input

        # 4. 执行工具
        if action in TOOLS:
            tool_fn, _ = TOOLS[action]            # 取出对应函数
            observation = tool_fn(action_input)   # 真正"动手"执行
            print(f"Observation: {observation}")
            # 把LLM的输出和观察结果都塞回对话，供下一轮推理
            messages.append({"role": "assistant", "content": output})
            messages.append({"role": "user", "content": f"Observation: {observation}"})
        else:
            # 工具名不存在时提示LLM重新选择
            messages.append({"role": "assistant", "content": output})
            messages.append({"role": "user", "content": f"Observation: 工具'{action}'不存在，请从可用工具中选择。"})

    return "达到最大步数，未能得出答案。"

if __name__ == "__main__":
    # 测试：计算并搜索的组合任务
    answer = react_agent("北京今天多少度？如果把这个温度乘以2再加10是多少？")
    print(f"\n最终答案：{answer}")

 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

# file: level3_plan_execute.py
# Plan-and-Execute架构智能体
import os
import json
from openai import OpenAI

client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

PLANNER_PROMPT = """你是一个任务规划器。请将用户任务拆分为有序的子任务列表。
每个子任务应可独立执行，粒度适中（不多于5个子任务）。
仅输出JSON数组，不要其他文字，格式如：
["子任务1", "子任务2"]

用户任务：{task}
"""

EXECUTOR_PROMPT = """你是一个任务执行器。请完成以下子任务。
你可以使用如下已知能力：搜索、计算、翻译、写作。
若需要外部信息，可基于常识合理假设并在结果中标注"假设"。
仅输出执行结果，不要解释过程。

子任务：{subtask}
上下文（已完成结果）：{context}
"""

def plan(task: str) -> list:
    """生成子任务计划：让LLM输出JSON数组"""
    resp = client.chat.completions.create(
        model="gpt-5.4",
        messages=[{"role": "user", "content": PLANNER_PROMPT.format(task=task)}],
        temperature=0.2,  # 规划阶段低温度，让拆解更稳定
    )
    raw = resp.choices[0].message.content.strip()
    try:
        return json.loads(raw)  # 把JSON文本解析成Python列表
    except json.JSONDecodeError:
        # 解析失败则退化为单任务，保证不崩
        return [task]

def execute(subtask: str, context: str) -> str:
    """执行单个子任务，带上已完成结果作为上下文"""
    resp = client.chat.completions.create(
        model="gpt-5.4",
        messages=[{"role": "user", "content": EXECUTOR_PROMPT.format(
            subtask=subtask, context=context
        )}],
        temperature=0.3,
    )
    return resp.choices[0].message.content.strip()

def plan_execute_agent(task: str) -> str:
    """Plan-and-Execute主流程：规划→执行→汇总"""
    # 1. 规划：拆出子任务清单
    subtasks = plan(task)
    print(f"计划：{subtasks}")

    # 2. 依次执行，并累积上下文（后面的子任务能看到前面的结果）
    context_parts = []
    for i, st in enumerate(subtasks, 1):
        result = execute(st, "\n".join(context_parts))
        context_parts.append(f"[{i}] {st} -> {result}")  # 拼进上下文
        print(f"执行[{i}]：{st}\n结果：{result}\n")

    # 3. 汇总：让LLM把所有子任务结果整理成最终答案
    summary_prompt = f"请根据以下子任务结果，回答原始任务：\n原始任务：{task}\n子任务结果：\n" + "\n".join(context_parts)
    resp = client.chat.completions.create(
        model="gpt-5.4",
        messages=[{"role": "user", "content": summary_prompt}],
        temperature=0.3,
    )
    return resp.choices[0].message.content.strip()

if __name__ == "__main__":
    print(plan_execute_agent("帮我调研Python和Rust在Web开发中的优劣，并给出学习建议。"))

 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

# file: level4_reflection.py
# Reflection架构智能体：代码生成与自我改进
import os
from openai import OpenAI

client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

GENERATOR_PROMPT = """你是一个Python工程师。请根据需求生成代码。
若收到改进建议，请据此修订代码并输出完整版本。

需求：{requirement}
上一版代码：
{previous_code}
改进建议：
{feedback}

只输出Python代码，不要解释。
"""

CRITIC_PROMPT = """你是一个严格的代码审查员。请审查以下代码：
需求：{requirement}
代码：
{code}

请从以下维度评审：正确性、健壮性、可读性、边界处理。
若代码已合格，输出：PASS
否则输出具体改进建议（不超过3条）。
"""

def generate(requirement: str, previous_code: str, feedback: str) -> str:
    """生成器：产出或修订代码"""
    resp = client.chat.completions.create(
        model="gpt-5.4",
        messages=[{"role": "user", "content": GENERATOR_PROMPT.format(
            requirement=requirement,
            previous_code=previous_code or "(无)",  # 首轮没有上一版
            feedback=feedback or "(无)",            # 首轮没有反馈
        )}],
        temperature=0.4,  # 略高温度，鼓励多样化尝试
    )
    return resp.choices[0].message.content.strip()

def critique(requirement: str, code: str) -> str:
    """评审器：检查并给出反馈（合格返回PASS）"""
    resp = client.chat.completions.create(
        model="gpt-5.4",
        messages=[{"role": "user", "content": CRITIC_PROMPT.format(
            requirement=requirement, code=code
        )}],
        temperature=0.2,  # 审查要严格稳定，用低温度
    )
    return resp.choices[0].message.content.strip()

def reflection_agent(requirement: str, max_iterations: int = 3) -> str:
    """Reflection主循环：生成→评审→改进，直到通过或达上限"""
    code = ""        # 上一版代码，首轮为空
    feedback = ""    # 评审反馈，首轮为空
    for i in range(max_iterations):
        # 1. 生成/修订代码
        code = generate(requirement, code, feedback)
        print(f"\n--- 第{i+1}轮代码 ---\n{code}\n")

        # 2. 评审
        feedback = critique(requirement, code)
        print(f"评审反馈：{feedback}\n")

        # 3. 通过则退出循环
        if "PASS" in feedback:
            print("代码通过审查。")
            break
    return code

if __name__ == "__main__":
    req = "写一个函数，输入整数列表，返回其中第二大的数。要求处理空列表、单元素列表等边界情况。"
    final_code = reflection_agent(req)
    print(f"\n最终代码：\n{final_code}")