《Python大模型Agent开发工程化实战：从ReAct循环到自主多智能体系统》

import asyncio
from typing import List, Dict
from pydantic import BaseModel, Field

class AgentState(BaseModel):
    messages: List[Dict] = Field(default_factory=list)
    scratchpad: str = Field(default="")  # 记录推理过程
    step_count: int = Field(default=0)
    max_steps: int = Field(default=10)

class EnhancedReActAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        
    async def arun(self, query: str) -> str:
        state = AgentState(messages=[{"role": "user", "content": query}])
        
        while state.step_count < state.max_steps:
            # 1. 推理阶段：生成Thought和Action
            response = await self.llm.agenerate(
                self._build_prompt(state)
            )
            thought, action_name, action_input = self._parse_llm_output(response)
            
            # 2. 行动阶段：执行工具（带超时控制）
            try:
                observation = await asyncio.wait_for(
                    self._execute_tool(action_name, action_input),
                    timeout=30.0
                )
            except asyncio.TimeoutError:
                observation = "Error: Tool execution timed out."
            
            # 3. 观察阶段：将结果追加到上下文中
            state.messages.append({"role": "assistant", "content": f"Observation: {observation}"})
            state.scratchpad += f"\n{response}\nObservation: {observation}"
            state.step_count += 1
            
            # 4. 提前终止判定：若模型输出Final Answer
            if self._is_final_answer(response):
                return self._extract_final_answer(response)
                
        return "Max steps exceeded. Please refine your request."

class MessageBus:
    def __init__(self):
        self.queues: Dict[str, asyncio.Queue] = {}
    
    def register(self, agent_name: str):
        self.queues[agent_name] = asyncio.Queue()
    
    async def send(self, to: str, message: Dict):
        await self.queues[to].put(message)

# 定义Worker Agent的基类
class BaseAgent:
    def __init__(self, name, bus):
        self.name = name
        self.bus = bus
        self.bus.register(name)
        
    async def run(self):
        while True:
            msg = await self.bus.queues[self.name].get()
            if msg.get("type") == "shutdown":
                break
            # 执行特定任务并将结果发回主控Agent
            result = await self.process(msg["payload"])
            await self.bus.send("orchestrator", {"sender": self.name, "result": result})

from pydantic import BaseModel, Field
from typing import Optional

class LinuxCommandInput(BaseModel):
    command: str = Field(description="Linux shell command to execute")
    timeout: Optional[int] = Field(default=30, description="Timeout in seconds")
    safe_mode: bool = Field(default=True, description="If True, only read-only commands are allowed")

class LinuxCommandTool:
    name = "execute_linux_command"
    description = "Execute shell commands on Linux system with safety checks"
    args_schema = LinuxCommandInput
    
    async def _arun(self, command: str, timeout: int = 30, safe_mode: bool = True):
        # 实现：白名单过滤（仅允许 ls, cat, ps 等安全命令）
        if safe_mode and not self._is_command_safe(command):
            return "Safety policy violation: Command blocked."
        # ...异步执行逻辑

async def execute_parallel_tools(tool_calls: List[Dict]):
    tasks = []
    for call in tool_calls:
        tool = tools_registry[call["name"]]
        tasks.append(tool._arun(**call["args"]))
    return await asyncio.gather(*tasks, return_exceptions=True)