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

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

    def process(self, raw_input):
        """
        处理原始输入
        - 用户消息
        - 工具返回结果
        - 系统状态
        """
        if isinstance(raw_input, dict):  # isinstance检查类型
            # 结构化数据（工具返回）
            return self._format_observation(raw_input)
        else:
            # 文本输入
            return raw_input

    def _format_observation(self, data):
        """将结构化数据格式化为文本"""
        return json.dumps(data, ensure_ascii=False, indent=2)  # json.dumps将Python对象→JSON字符串

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):
        """基于上下文进行推理"""
        response = self.llm.generate(
            system_prompt=self.system_prompt,
            context=context
        )
        return self._parse_response(response)

    def _parse_response(self, response):
        """解析LLM输出，提取Thought和Action"""
        # 解析逻辑...
        pass

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

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

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

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

        try:  # try/except捕获异常，防止程序崩溃
            result = self.tools[tool_name]['function'](**params)
            return result
        except Exception as e:
            return f"Error: {str(e)}"

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

    def __init__(self):
        # 短期记忆（当前会话）
        self.short_term = []

        # 长期记忆（向量数据库）
        self.long_term = VectorDB()

    def add_short_term(self, item):
        """添加到短期记忆"""
        self.short_term.append(item)

        # 保持最近N条
        if len(self.short_term) > 10:
            self.short_term.pop(0)

    def add_long_term(self, item, embedding):
        """添加到长期记忆"""
        self.long_term.store(item, embedding)

    def retrieve(self, query_embedding, k=5):
        """检索相关记忆"""
        # 从长期记忆中检索
        relevant = self.long_term.search(query_embedding, k)

        # 结合短期记忆
        return self.short_term + relevant

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

# OpenAI Function Calling示例
import openai

functions = [
    {
        "name": "get_weather",
        "description": "获取天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        }
    }
]

from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "北京今天天气怎么样？"}
    ],
    tools=[{"type": "function", "function": f} for f in functions],
    tool_choice="auto"
)

# 模型决定调用函数
tool_calls = response.choices[0].message.tool_calls
if tool_calls:
    tool_call = tool_calls[0]
    function_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)  # json.loads将JSON字符串→Python对象

    # 执行函数
    result = get_weather(**arguments)

    # 将结果返回给模型
    second_response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "user", "content": "北京今天天气怎么样？"},
            response.choices[0].message,
            {"role": "tool", "tool_call_id": tool_call.id, "content": result}
        ]
    )

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

    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)
            else:
                break

        return result

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

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

        reflection = self.llm.generate(reflection_prompt)
        return self.parse_reflection(reflection)

class Planner:
    """任务规划器"""

    def create_plan(self, goal):
        """创建实现目标的计划"""

        planning_prompt = f"""
        目标：{goal}

        请制定一个详细的执行计划：
        1. 将目标分解为具体步骤
        2. 每个步骤应该是可执行的
        3. 明确步骤之间的依赖关系
        4. 考虑可能的错误和备选方案

        格式：
        Step 1: [描述] (依赖: 无)
        Step 2: [描述] (依赖: Step 1)
        ...
        """

        plan_text = self.llm.generate(planning_prompt)
        return self.parse_plan(plan_text)

    def execute_plan(self, plan):
        """执行计划"""
        results = {}

        for step in plan.steps:
            # 检查依赖是否满足
            if not all(dep in results for dep in step.dependencies):  # all()全部为True才返回True
                continue

            # 执行步骤
            context = {dep: results[dep] for dep in step.dependencies}
            result = self.execute_step(step, context)
            results[step.id] = result

            # 检查是否需要调整计划
            if self.should_replan(result):
                plan = self.replan(plan, results)

        return results

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.now()

class CommunicationProtocol:
    """通信协议"""

    def request(self, sender, receiver, task):
        """请求协助"""
        return Message(sender, receiver, "REQUEST", task)

    def inform(self, sender, receiver, information):
        """通知信息"""
        return Message(sender, receiver, "INFORM", information)

    def propose(self, sender, receiver, proposal):
        """提出建议"""
        return Message(sender, receiver, "PROPOSE", proposal)

class Negotiation:
    """协商机制"""

    def negotiate(self, agent_a, agent_b, issue):
        """
        双方就某个问题进行协商
        """
        # Agent A提出建议
        proposal_a = agent_a.make_proposal(issue)

        # Agent B评估
        evaluation = agent_b.evaluate(proposal_a)

        if evaluation.acceptable:
            return self.agreement(proposal_a)

        # Agent B提出反建议
        counter_proposal = agent_b.make_counter_proposal(issue, proposal_a)

        # 多轮协商...
        return self.iterative_negotiation(agent_a, agent_b, issue)

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

    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.memory = []

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

        for step in range(max_steps):
            # 构建prompt
            prompt = self._build_prompt()

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

            # 解析Thought和Action
            thought, action, action_input = self._parse(response)

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

            if action == "finish":
                return action_input

            # 执行工具
            observation = self.tools[action](**action_input)
            print(f"Observation: {observation}\n")

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

        return "Max steps reached"

from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_openai import ChatOpenAI
from langchain import hub

# 定义工具
search = DuckDuckGoSearchRun()
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="用于搜索实时信息"
    )
]

# 加载预定义的ReAct prompt
prompt = hub.pull("hwchase17/react")

# 创建LLM
llm = ChatOpenAI(model="gpt-4o")

# 创建Agent
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行
response = agent_executor.invoke({
    "input": "2024年诺贝尔物理学奖得主是谁？"
})
print(response)

class MultiAgentSystem:
    """多智能体系统示例"""

    def __init__(self):
        self.agents = {
            'researcher': ResearchAgent(),
            'writer': WriterAgent(),
            'reviewer': ReviewerAgent()
        }

    def collaborative_writing(self, topic):
        """协作写作"""
        # 研究员收集资料
        research = self.agents['researcher'].research(topic)

        # 作者撰写初稿
        draft = self.agents['writer'].write(topic, research)

        # 审阅者评审
        feedback = self.agents['reviewer'].review(draft)

        # 根据反馈修改（迭代）
        while not feedback.approved:
            draft = self.agents['writer'].revise(draft, feedback)
            feedback = self.agents['reviewer'].review(draft)

        return draft