02. 智能体系统（Agent Systems）¶

import json

class PerceptionModule:
    """感知模块：处理环境输入"""

    def process(self, raw_input):
        """
        处理原始输入
        - 用户消息（文本）
        - 工具返回结果（结构化数据）
        - 系统状态（事件通知）
        """
        if isinstance(raw_input, dict):
            # 结构化数据（工具返回）→ 格式化为文本
            return self._format_observation(raw_input)
        else:
            # 文本输入 → 直接传递
            return raw_input

    def _format_observation(self, data):
        """将结构化数据格式化为可读文本"""
        return json.dumps(data, ensure_ascii=False, indent=2)

import re

class ReasoningEngine:
    """推理引擎：基于 LLM 进行思考和决策"""

    def __init__(self, llm):
        self.llm = llm
        self.system_prompt = """你是一个智能助手。你可以使用以下工具：

可用工具：
{tools_description}

请按以下格式思考：
Thought: 你的思考过程
Action: 工具名称
Action Input: 工具参数（JSON格式）

或者完成任务：
Thought: 任务完成
Final Answer: 最终答案
"""

    def think(self, context, tools_description=""):
        """基于上下文进行推理"""
        prompt = self.system_prompt.format(tools_description=tools_description)
        response = self.llm.generate(
            system_prompt=prompt,
            context=context
        )
        return self._parse_response(response)

    def _parse_response(self, response):
        """解析 LLM 输出，提取 Thought 和 Action"""
        result = {"thought": "", "action": None, "action_input": {}, "final_answer": None}

        # 提取 Thought
        thought_match = re.search(r'Thought:\s*(.+?)(?=\n(?:Action|Final)|$)', response, re.DOTALL)
        if thought_match:
            result["thought"] = thought_match.group(1).strip()

        # 提取 Action
        action_match = re.search(r'Action:\s*(\w+)', response)
        if action_match:
            result["action"] = action_match.group(1)

        # 提取 Action Input
        input_match = re.search(r'Action Input:\s*(\{.+?\})', response, re.DOTALL)
        if input_match:
            result["action_input"] = json.loads(input_match.group(1))

        # 提取 Final Answer
        answer_match = re.search(r'Final Answer:\s*(.+?)$', response, re.DOTALL)
        if answer_match:
            result["final_answer"] = answer_match.group(1).strip()

        return result

class ToolExecutor:
    """工具执行器：管理和执行工具"""

    def __init__(self):
        self.tools = {}

    def register_tool(self, name, func, description, parameters):
        """注册工具"""
        self.tools[name] = {
            'function': func,
            'description': description,
            'parameters': parameters
        }

    def execute(self, tool_name, params):
        """执行工具"""
        if tool_name not in self.tools:
            return f"Error: Tool '{tool_name}' not found"

        try:
            result = self.tools[tool_name]['function'](**params)
            return result
        except TypeError as e:
            return f"Error: 参数错误 - {str(e)}"
        except Exception as e:
            return f"Error: 执行失败 - {str(e)}"

    def get_tools_description(self):
        """获取所有工具的描述（用于构建 prompt）"""
        descriptions = []
        for name, tool in self.tools.items():
            descriptions.append(f"- {name}: {tool['description']}")
        return "\n".join(descriptions)

class MemorySystem:
    """记忆系统：管理短期和长期记忆"""

    def __init__(self, max_short_term=10):
        # 短期记忆（当前会话的最近 N 条）
        self.short_term = []
        self.max_short_term = max_short_term

        # 长期记忆（向量数据库，实际使用 ChromaDB / FAISS 等）
        # self.long_term = VectorDB()

    def add_short_term(self, item):
        """添加到短期记忆"""
        self.short_term.append(item)
        # FIFO：保持最近 N 条
        if len(self.short_term) > self.max_short_term:
            self.short_term.pop(0)

    def get_context(self):
        """获取当前上下文（用于构建 prompt）"""
        return "\n".join(self.short_term)

    def clear_short_term(self):
        """清空短期记忆（新会话开始时）"""
        self.short_term = []

# 工具的标准定义（OpenAI Function Calling 格式）
tool_definition = {
    "name": "get_weather",
    "description": "获取指定城市的天气信息",
    "parameters": {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "城市名称，如'北京'"
            },
            "date": {
                "type": "string",
                "description": "日期，如'今天'、'明天'或'2026-03-25'"
            }
        },
        "required": ["location"]
    }
}

"""
OpenAI Function Calling 完整流程
展示了从工具定义到执行再到结果返回的完整循环
"""
import json
from openai import OpenAI

client = OpenAI()

# Step 1: 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "城市名称"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["location"]
            }
        }
    }
]

# Step 2: 发送请求（模型决定是否调用工具）
response = client.chat.completions.create(
    model="gpt-5-mini",
    messages=[
        {"role": "user", "content": "北京今天天气怎么样？"}
    ],
    tools=tools,
    tool_choice="auto"  # auto: 模型自动决定；none: 不调用；required: 必须调用
)

message = response.choices[0].message

# Step 3: 处理工具调用
if message.tool_calls:
    tool_call = message.tool_calls[0]
    function_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)

    # 执行实际函数
    def get_weather(location, unit="celsius"):
        # 实际应用中调用天气 API
        return json.dumps({
            "location": location,
            "temperature": "15°C" if unit == "celsius" else "59°F",
            "weather": "晴",
            "wind": "微风"
        }, ensure_ascii=False)

    result = get_weather(**arguments)

    # Step 4: 将工具结果返回给模型
    second_response = client.chat.completions.create(
        model="gpt-5-mini",
        messages=[
            {"role": "user", "content": "北京今天天气怎么样？"},
            message,  # 模型的工具调用消息
            {"role": "tool", "tool_call_id": tool_call.id, "content": result}
        ]
    )

    print(second_response.choices[0].message.content)
else:
    # 模型直接回答（无需工具）
    print(message.content)

class ReflectiveAgent:
    """具备反思能力的智能体"""

    def __init__(self, llm):
        self.llm = llm

    def execute_with_reflection(self, task, max_retries=3):
        """执行任务，支持反思和重试"""

        for attempt in range(max_retries):
            # 执行任务
            result = self.execute(task)

            # 反思执行结果
            reflection = self.reflect(task, result)

            if reflection["is_success"]:
                return result

            if reflection["should_retry"]:
                # 根据反馈调整策略
                task = self.adjust_strategy(task, reflection["feedback"])
                print(f"重试 {attempt + 1}/{max_retries}，调整：{reflection['feedback']}")
            else:
                break

        return result

    def reflect(self, task, result):
        """反思执行结果"""
        reflection_prompt = f"""
任务：{task}
执行结果：{result}

请反思：
1. 结果是否正确？
2. 过程中是否有错误？
3. 如何改进？
4. 是否需要重试？

请以 JSON 格式输出：
{{"is_success": bool, "should_retry": bool, "feedback": "改进建议"}}
"""
        reflection_text = self.llm.generate(reflection_prompt)
        return self.parse_reflection(reflection_text)

from pydantic import BaseModel
from typing import Optional
from enum import Enum

class StepStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

class PlanStep(BaseModel):
    """结构化计划中的单个步骤"""
    step_id: int
    action: str               # 要执行的动作描述
    tool: Optional[str]       # 使用的工具名称
    parameters: dict          # 工具参数
    depends_on: list[int]     # 依赖的前置步骤 ID
    expected_output: str      # 预期输出描述

class StructuredPlan(BaseModel):
    """结构化执行计划"""
    goal: str
    steps: list[PlanStep]
    estimated_steps: int
    fallback_strategy: str    # 如果计划失败的回退策略

# 使用 LLM 生成结构化计划
STRUCTURED_PLANNING_PROMPT = """
你是一个任务规划专家。请将用户的目标分解为可执行的结构化计划。

用户目标：{goal}

可用工具：
{available_tools}

请严格按照以下 JSON Schema 输出计划：
{schema}

要求：
1. 每个步骤必须指定明确的工具和参数
2. 使用 depends_on 字段表示步骤间的依赖关系
3. 提供回退策略以应对可能的失败
"""

async def generate_structured_plan(
    goal: str,
    tools: list[dict],
    llm_client
) -> StructuredPlan:
    """使用 LLM 生成结构化执行计划"""
    schema = StructuredPlan.model_json_schema()
    prompt = STRUCTURED_PLANNING_PROMPT.format(
        goal=goal,
        available_tools="\n".join(f"- {t['name']}: {t['description']}" for t in tools),
        schema=schema
    )

    response = await llm_client.chat.completions.create(
        model="gpt-5-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"}  # 强制 JSON 输出
    )

    plan = StructuredPlan.model_validate_json(response.choices[0].message.content)
    return plan

import subprocess
from pathlib import Path

class CodeAgent:
    """
    代码执行 Agent：生成 Python 代码并在沙箱中执行。
    参考 HuggingFace smolagents 的 CodeAgent 设计。
    """

    def __init__(self, llm_client, allowed_modules=None):
        self.llm_client = llm_client
        # 白名单机制：只允许导入安全的模块
        self.allowed_modules = allowed_modules or [
            "math", "json", "re", "datetime", "collections",
            "statistics", "itertools", "string", "random"
        ]

    async def run(self, question: str) -> dict:
        """生成代码并执行"""
        # Step 1: 生成代码
        prompt = f"""你是一个代码执行 Agent。请编写 Python 代码来解决用户的问题。

规则：
1. 只能使用以下模块：{', '.join(self.allowed_modules)}
2. 将最终结果赋值给变量 `result`
3. 不要使用文件操作、网络请求或系统调用
4. 代码必须能在受限环境中安全执行

用户问题：{question}

请直接输出可执行的 Python 代码，不要包含 markdown 标记。"""

        response = await self.llm_client.chat.completions.create(
            model="gpt-5-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0  # 确定性输出
        )
        code = response.choices[0].message.content.strip()

        # Step 2: 安全检查
        if not self._safety_check(code):
            return {"success": False, "error": "代码未通过安全检查"}

        # Step 3: 沙箱执行
        result = self._execute_in_sandbox(code)
        return result

    def _safety_check(self, code: str) -> bool:
        """基础安全检查"""
        dangerous_patterns = [
            "import os", "import sys", "import subprocess",
            "open(", "exec(", "eval(", "__import__",
            "rm -rf", "shutil.rmtree"
        ]
        return not any(pattern in code for pattern in dangerous_patterns)

    def _execute_in_sandbox(self, code: str, timeout: int = 30) -> dict:
        """在受限环境中执行代码"""
        try:
            result = subprocess.run(
                ["python", "-c", code],
                capture_output=True,
                text=True,
                timeout=timeout
            )
            if result.returncode == 0:
                return {"success": True, "output": result.stdout.strip(), "code": code}
            else:
                return {"success": False, "error": result.stderr.strip(), "code": code}
        except subprocess.TimeoutExpired:
            return {"success": False, "error": f"执行超时（{timeout}s）"}
        except Exception as e:
            return {"success": False, "error": str(e)}

import time

class Message:
    """智能体间消息"""

    def __init__(self, sender, receiver, msg_type, content):
        self.sender = sender
        self.receiver = receiver
        self.type = msg_type  # REQUEST, INFORM, PROPOSE, ACCEPT, REJECT
        self.content = content
        self.timestamp = time.time()

class CommunicationBus:
    """通信总线：管理智能体间的消息传递"""

    def __init__(self):
        self.message_queue = []

    def send(self, message: Message):
        """发送消息"""
        self.message_queue.append(message)

    def receive(self, agent_name: str) -> list[Message]:
        """接收指定 Agent 的消息"""
        messages = [
            m for m in self.message_queue
            if m.receiver == agent_name
        ]
        # 从队列中移除已接收的消息
        for m in messages:
            self.message_queue.remove(m)
        return messages

"""
完整的 ReAct 智能体实现
包含工具注册、推理循环和记忆管理
"""
import json
import re

class SimpleReActAgent:
    """简单的 ReAct 智能体实现"""

    def __init__(self, llm, tools=None):
        self.llm = llm
        self.tools = tools or {}
        self.memory = []

    def register_tool(self, name, func, description):
        """注册工具"""
        self.tools[name] = {"function": func, "description": description}

    def run(self, query, max_steps=10):
        """运行智能体"""
        self.memory = [f"User: {query}"]

        for step in range(max_steps):
            # 构建 prompt
            tools_desc = "\n".join(
                f"- {name}: {info['description']}"
                for name, info in self.tools.items()
            )
            prompt = f"""可用工具：
{tools_desc}

历史记录：
{chr(10).join(self.memory)}

请按格式回答：
Thought: 你的思考
Action: 工具名 或 finish
Action Input: {{"key": "value"}} 或 最终答案"""

            # 获取模型输出
            response = self.llm.generate(prompt)

            # 解析 Thought 和 Action
            thought = self._extract(response, r'Thought:\s*(.+?)(?=\n)')
            action = self._extract(response, r'Action:\s*(\w+)')
            action_input = self._extract(response, r'Action Input:\s*(.+?)$')

            print(f"Step {step + 1}:")
            print(f"  Thought: {thought}")
            print(f"  Action: {action}")

            if action == "finish":
                return action_input

            # 执行工具
            if action in self.tools:
                try:
                    params = json.loads(action_input) if action_input else {}
                    observation = self.tools[action]["function"](**params)
                except Exception as e:
                    observation = f"Error: {str(e)}"
            else:
                observation = f"Error: Unknown tool '{action}'"

            print(f"  Observation: {observation}\n")

            # 添加到记忆
            self.memory.extend([
                f"Thought: {thought}",
                f"Action: {action}({action_input})",
                f"Observation: {observation}"
            ])

        return "Max steps reached"

    def _extract(self, text, pattern):
        """从文本中提取匹配内容"""
        match = re.search(pattern, text, re.DOTALL)
        return match.group(1).strip() if match else ""


# 使用示例
if __name__ == "__main__":
    # 模拟 LLM（实际使用 OpenAI / Anthropic API）
    class MockLLM:
        def generate(self, prompt):
            if "天气" in prompt or "weather" in prompt:
                return 'Thought: 用户想查天气\nAction: get_weather\nAction Input: {"location": "北京"}'
            return 'Thought: 任务完成\nAction: finish\nAction Input: 北京今天晴天，15°C'

    agent = SimpleReActAgent(MockLLM())
    agent.register_tool("get_weather", lambda location: "晴天，15°C", "获取天气信息")
    result = agent.run("北京天气怎么样？")
    print(f"最终答案: {result}")

"""
使用 LangGraph 构建生产级 ReAct Agent
LangGraph 是 LangChain Agent 的推荐替代方案
"""
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 定义工具（使用 @tool 装饰器）
@tool
def search(query: str) -> str:
    """搜索实时信息"""
    # 实际使用 DuckDuckGoSearchRun / Tavily 等
    return f"搜索结果：{query} 的相关信息..."

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)  # 生产环境建议用 sympy
        return f"计算结果：{expression} = {result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# 创建 LLM
llm = ChatOpenAI(model="gpt-5-mini")

# 创建 Agent（LangGraph 方式）
agent = create_react_agent(
    model=llm,
    tools=[search, calculator]
)

# 运行
response = agent.invoke({
    "messages": [{"role": "user", "content": "2024 年诺贝尔物理学奖得主是谁？"}]
})
print(response["messages"][-1].content)

"""
多智能体协作系统示例
使用协调者模式分配任务
"""
from enum import Enum
from typing import Optional

class AgentRole(str, Enum):
    RESEARCHER = "researcher"
    WRITER = "writer"
    REVIEWER = "reviewer"

class CollaborativeSystem:
    """多智能体协作系统"""

    def __init__(self, llm):
        self.llm = llm
        self.agents = {
            AgentRole.RESEARCHER: "你是一个研究专家，负责收集和整理信息。",
            AgentRole.WRITER: "你是一个写作专家，负责根据资料撰写文章。",
            AgentRole.REVIEWER: "你是一个审阅专家，负责评审文章质量并提出修改建议。",
        }

    def run(self, topic: str, max_iterations: int = 3) -> str:
        """协作写作流程"""
        # Phase 1: 研究
        research = self._call_agent(
            AgentRole.RESEARCHER,
            f"请研究以下主题并整理关键信息：{topic}"
        )

        # Phase 2: 写作
        draft = self._call_agent(
            AgentRole.WRITER,
            f"基于以下资料撰写文章：\n资料：{research}\n主题：{topic}"
        )

        # Phase 3: 审阅 + 迭代修改
        for i in range(max_iterations):
            review = self._call_agent(
                AgentRole.REVIEWER,
                f"请审阅以下文章，指出问题并给出修改建议：\n{draft}\n\n如果质量合格，请回复 'APPROVED'。"
            )

            if "APPROVED" in review:
                print(f"审阅通过（第 {i + 1} 轮）")
                return draft

            # 根据反馈修改
            draft = self._call_agent(
                AgentRole.WRITER,
                f"请根据审阅意见修改文章：\n原文：{draft}\n审阅意见：{review}"
            )

        return draft

    def _call_agent(self, role: AgentRole, prompt: str) -> str:
        """调用指定角色的 Agent"""
        system_prompt = self.agents[role]
        response = self.llm.generate(f"{system_prompt}\n\n{prompt}")
        return response