Agentic Workflow架构演进：从Chatbot到多智能体协作的实战指南

from pydantic import BaseModel, Field
from typing import List, Optional

class TaskDefinition(BaseModel):
    """结构化任务定义"""
    objective: str = Field(description="任务目标，一句话描述")
    constraints: List[str] = Field(description="约束条件列表")
    success_criteria: List[str] = Field(description="成功验证标准")
    max_iterations: int = Field(default=5, description="最大迭代次数")
    fallback_strategy: str = Field(default="ask_human", description="回退策略")
    tools_allowed: List[str] = Field(description="允许使用的工具列表")
    output_format: str = Field(description="输出格式规范")

class AgentConfig(BaseModel):
    """智能体配置"""
    name: str
    role: str
    model: str = "gpt-6-astra"
    system_prompt: str
    tools: List[str]
    max_tokens: int = 8192
    temperature: float = 0.3
    memory_scope: str = "task"  # task | session | global

class SequentialPipeline:
    """顺序流水线编排器"""
    def __init__(self, agents: List[Agent]):
        self.agents = agents
        self.state = {}

    async def execute(self, input_data: str) -> dict:
        result = {"input": input_data, "steps": []}
        current_data = input_data

        for i, agent in enumerate(self.agents):
            step_result = await agent.run(
                input_data=current_data,
                context=self.state
            )

            # 验证输出
            if not self._validate(step_result, agent):
                raise AgentOutputError(
                    f"Agent {agent.name} output validation failed at step {i}"
                )

            current_data = step_result.output
            self.state[f"step_{i}_result"] = step_result
            result["steps"].append({
                "agent": agent.name,
                "status": "success",
                "output": step_result.output
            })

        result["final_output"] = current_data
        return result

class AgentMemorySystem:
    """多智能体共享记忆系统"""

    def __init__(self, redis_client, vector_db):
        self.redis = redis_client  # 短期记忆
        self.vector_db = vector_db  # 语义记忆
        self.session_id = str(uuid.uuid4())

    async def store(self, key: str, value: str, scope: str = "task"):
        """存储记忆"""
        if scope == "task":
            # 短期记忆：存储在Redis，设置TTL
            await self.redis.setex(
                f"memory:{self.session_id}:{key}",
                3600,  # 1小时过期
                json.dumps(value)
            )
        elif scope == "semantic":
            # 语义记忆：存入向量数据库
            embedding = await self.get_embedding(value)
            await self.vector_db.upsert(
                id=f"{self.session_id}:{key}",
                embedding=embedding,
                metadata={"content": value, "session": self.session_id}
            )

    async def retrieve(self, query: str, top_k: int = 5) -> list:
        """检索相关记忆"""
        # 从短期记忆中获取当前会话上下文
        short_term = await self.redis.get(f"memory:{self.session_id}:context")

        # 从语义记忆中进行相似性搜索
        query_embedding = await self.get_embedding(query)
        long_term = await self.vector_db.search(
            vector=query_embedding,
            top_k=top_k,
            filter={"session": self.session_id}
        )

        return {
            "short_term": json.loads(short_term) if short_term else None,
            "long_term": long_term
        }

import subprocess
import tempfile
import shutil

class SecureSandbox:
    """Agent代码执行安全沙箱"""

    def __init__(self):
        self.workspace = tempfile.mkdtemp(prefix="sandbox_")

    def execute_code(self, code: str, language: str = "python") -> dict:
        """在隔离环境中执行代码"""
        if language == "python":
            return self._execute_python(code)
        elif language == "javascript":
            return self._execute_js(code)
        else:
            raise ValueError(f"Unsupported language: {language}")

    def _execute_python(self, code: str) -> dict:
        # 写入临时文件
        script_path = os.path.join(self.workspace, "script.py")
        with open(script_path, "w") as f:
            f.write(code)

        # 在Docker容器中执行
        result = subprocess.run(
            [
                "docker", "run", "--rm",
                "--network=none",  # 禁止网络访问
                "--memory=512m",   # 限制内存
                "--cpus=1",        # 限制CPU
                "--read-only",     # 只读文件系统
                "--tmpfs=/tmp:rw,size=64m",  # 临时写入空间
                "-v", f"{self.workspace}:/workspace:ro",
                "python:3.12-slim",
                "python", "/workspace/script.py"
            ],
            capture_output=True,
            text=True,
            timeout=30  # 超时限制
        )

        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "returncode": result.returncode
        }

    def cleanup(self):
        """清理沙箱环境"""
        shutil.rmtree(self.workspace, ignore_errors=True)

import time
from dataclasses import dataclass
from collections import defaultdict

@dataclass
class AgentMetrics:
    """Agent执行指标收集器"""
    task_id: str
    start_time: float = time.time()
    iterations: int = 0
    tool_calls: defaultdict = defaultdict(int)
    token_usage: int = 0
    errors: list = None

    def record_iteration(self):
        self.iterations += 1

    def record_tool_call(self, tool_name: str):
        self.tool_calls[tool_name] += 1

    def record_tokens(self, count: int):
        self.token_usage += count

    def record_error(self, error: str):
        if self.errors is None:
            self.errors = []
        self.errors.append(error)

    def summary(self) -> dict:
        elapsed = time.time() - self.start_time
        return {
            "task_id": self.task_id,
            "duration_seconds": round(elapsed, 2),
            "iterations": self.iterations,
            "tool_calls": dict(self.tool_calls),
            "token_usage": self.token_usage,
            "errors": self.errors or [],
            "status": "failed" if self.errors else "success"
        }

async def automated_code_review(pr_url: str) -> dict:
    """自动化代码审查工作流"""
    # 1. Research Agent：收集PR信息和上下文
    pr_info = await research_agent.run(
        task=TaskDefinition(
            objective="收集PR信息、变更文件、提交历史和相关文档",
            tools_allowed=["github_api", "file_reader"],
            output_format="json"
        ),
        input_data=pr_url
    )

    # 2. Code Agent：分析代码变更
    analysis = await code_agent.run(
        task=TaskDefinition(
            objective="分析代码变更的功能、架构影响和潜在问题",
            constraints=["关注安全漏洞", "检查性能影响", "验证测试覆盖"],
            tools_allowed=["code_reader", "test_runner", "security_scanner"],
            output_format="structured_report"
        ),
        input_data=pr_info.output
    )

    # 3. Review Agent：生成审查意见
    review = await review_agent.run(
        task=TaskDefinition(
            objective="基于代码分析结果生成结构化审查报告",
            success_criteria=["每个问题有严重级别", "提供修复建议", "引用相关代码行"],
            max_iterations=3,
            output_format="markdown"
        ),
        input_data=analysis.output
    )

    return {
        "pr_url": pr_url,
        "review_report": review.output,
        "metrics": metrics.summary()
    }