AI Agent智能体架构设计：从理论到实践

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 传统Chatbot
chatbot_response = llm.generate("帮我查询天气")
# 单轮对话，无状态，无法执行操作

# AI Agent
agent = Agent(
    tools=[weather_api, calendar_api],
    memory=LongTermMemory(),
    planner=ReActPlanner()
)
result = agent.run("帮我查明天天气，如果有雨则安排线上会议")
# 多步推理，工具调用，状态管理

 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

class ReActPlanner:
    """Reasoning + Acting 规划器"""

    def plan(self, query: str, context: Context) -> Plan:
        thoughts = []

        while not self.is_complete(context):
            # 思考：分析当前状态
            thought = self.llm.generate(
                f"Query: {query}\n"
                f"Context: {context}\n"
                f"What should I do next?"
            )
            thoughts.append(thought)

            # 行动：选择并执行工具
            action = self.extract_action(thought)
            observation = self.execute_tool(action)
            context.add_observation(observation)

        return Plan(thoughts, context)

    def extract_action(self, thought: str) -> Action:
        # 解析LLM输出，提取工具调用
        return Action(
            tool=self.parse_tool_name(thought),
            parameters=self.parse_parameters(thought)
        )

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

class ChainOfThoughtPlanner:
    """思维链规划器"""

    def plan(self, query: str) -> Plan:
        prompt = f"""
        目标: {query}

        请按以下格式思考：

        思考步骤1: [分析问题]
        思考步骤2: [确定需要的信息]
        思考步骤3: [选择工具]
        思考步骤4: [执行方案]
        思考步骤5: [验证结果]

        让我们一步步思考。
        """

        response = self.llm.generate(prompt)

        return self.parse_chain(response)

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

class TODOPlanner:
    """任务分解规划器"""

    def plan(self, goal: str) -> Plan:
        # 生成TODO列表
        prompt = f"""
        目标: {goal}

        请将此目标分解为具体的任务列表：
        1. [任务1]
        2. [任务2]
        3. [任务3]
        ...

        对每个任务标记：
        - 待办 (TODO)
        - 进行中 (DOING)
        - 已完成 (DONE)
        """

        todos = self.llm.generate(prompt)
        return self.parse_todos(todos)

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

class WorkingMemory:
    """短期记忆，存储当前会话上下文"""

    def __init__(self, window_size: int = 10):
        self.messages = []
        self.window_size = window_size

    def add_message(self, role: str, content: str):
        self.messages.append({
            "role": role,
            "content": content,
            "timestamp": time.time()
        })

        # 滑动窗口
        if len(self.messages) > self.window_size:
            self.messages.pop(0)

    def get_context(self) -> str:
        return "\n".join([
            f"{msg['role']}: {msg['content']}"
            for msg in self.messages
        ])

 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

class LongTermMemory:
    """长期记忆，基于向量数据库"""

    def __init__(self, vector_db):
        self.db = vector_db
        self.embedding_model = OpenAIEmbedding()

    def store(self, content: str, metadata: dict):
        """存储记忆"""
        embedding = self.embedding_model.encode(content)

        self.db.insert(
            content=content,
            embedding=embedding,
            metadata=metadata
        )

    def retrieve(self, query: str, top_k: int = 5) -> list:
        """检索相关记忆"""
        query_embedding = self.embedding_model.encode(query)

        results = self.db.search(
            embedding=query_embedding,
            top_k=top_k
        )

        return results

    def reflect(self):
        """记忆反思和整理"""
        # 1. 识别重要记忆
        important = self.db.filter(
            metadata={"importance": {"$gt": 0.8}}
        )

        # 2. 总结和提炼
        summary = self.llm.generate(
            f"总结以下关键信息：\n{important}"
        )

        # 3. 存储提炼后的记忆
        self.store(summary, metadata={
            "type": "reflection",
            "timestamp": time.time()
        })

 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

import chromadb
from sentence_transformers import SentenceTransformer

class VectorMemoryStore:
    """向量记忆存储"""

    def __init__(self):
        self.client = chromadb.Client()
        self.collection = self.client.create_collection(
            name="agent_memories",
            metadata={"hnsw:space": "cosine"}
        )
        self.encoder = SentenceTransformer(
            'paraphrase-multilingual-MiniLM-L12-v2'
        )

    def add(self, memories: list[Memory]):
        """批量添加记忆"""
        embeddings = self.encoder.encode([
            m.content for m in memories
        ])

        self.collection.add(
            embeddings=embeddings.tolist(),
            documents=[m.content for m in memories],
            metadatas=[m.metadata for m in memories],
            ids=[m.id for m in memories]
        )

    def search(self, query: str, n_results: int = 5):
        """语义搜索"""
        query_embedding = self.encoder.encode([query])

        results = self.collection.query(
            query_embeddings=query_embedding.tolist(),
            n_results=n_results
        )

        return results

 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

from typing import Callable, Any
from pydantic import BaseModel

class Tool(BaseModel):
    name: str
    description: str
    parameters: dict
    function: Callable

class ToolRegistry:
    """工具注册表"""

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

    def register(self, tool: Tool):
        self.tools[tool.name] = tool

    def get_tool(self, name: str) -> Tool:
        return self.tools.get(name)

    def list_tools(self) -> list[Tool]:
        return list(self.tools.values())

# 注册常用工具
registry = ToolRegistry()

registry.register(Tool(
    name="search",
    description="搜索网络信息",
    parameters={
        "query": {
            "type": "string",
            "description": "搜索关键词"
        }
    },
    function=lambda query: google_search(query)
))

registry.register(Tool(
    name="calculator",
    description="执行数学计算",
    parameters={
        "expression": {
            "type": "string",
            "description": "数学表达式"
        }
    },
    function=lambda expr: eval(expr)
))

 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

class ToolExecutor:
    """工具执行器"""

    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    def execute(self, action: Action) -> Result:
        tool = self.registry.get_tool(action.tool_name)

        if not tool:
            return Result(error=f"Tool {action.tool_name} not found")

        try:
            # 参数验证
            self.validate_parameters(
                tool.parameters,
                action.parameters
            )

            # 执行工具
            result = tool.function(**action.parameters)

            return Result(data=result)

        except Exception as e:
            return Result(error=str(e))

    def validate_parameters(self, schema: dict, params: dict):
        """参数校验"""
        for key, value in schema.items():
            if key not in params:
                raise ValueError(f"Missing parameter: {key}")

            # 类型检查
            if value["type"] == "string":
                assert isinstance(params[key], str)
            elif value["type"] == "number":
                assert isinstance(params[key], (int, float))

 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

class OpenAIFunctionCalling:
    """OpenAI Function Calling 实现"""

    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)

    def execute_with_functions(
        self,
        prompt: str,
        functions: list[Tool]
    ) -> Any:
        # 构建function schemas
        function_schemas = [
            {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.parameters
            }
            for tool in functions
        ]

        # 调用LLM
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            functions=function_schemas,
            function_call="auto"
        )

        # 检查是否需要调用function
        if response.choices[0].message.function_call:
            function_call = response.choices[0].message.function_call

            # 执行function
            function_name = function_call.name
            arguments = json.loads(function_call.arguments)

            return self.execute_function(
                function_name,
                arguments,
                functions
            )

        return response.choices[0].message.content

 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

class Reflector:
    """Agent反思器"""

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

    def reflect(self, execution: Execution) -> Reflection:
        """反思执行过程"""
        prompt = f"""
        目标: {execution.goal}
        执行步骤: {execution.steps}
        结果: {execution.result}

        请分析：
        1. 是否达成目标？
        2. 哪些步骤做得好？
        3. 哪些步骤可以改进？
        4. 如果重来，会怎么做？
        """

        reflection = self.llm.generate(prompt)

        return Reflection(
            success=self.evaluate_success(execution),
            insights=self.extract_insights(reflection),
            improvements=self.suggest_improvements(reflection)
        )

    def evaluate_success(self, execution: Execution) -> bool:
        """评估是否成功"""
        # 使用LLM评估
        prompt = f"""
        目标: {execution.goal}
        结果: {execution.result}

        这个结果是否达成了目标？请回答是或否，并说明理由。
        """

        response = self.llm.generate(prompt)
        return "是" in response

 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

class Message:
    """Agent消息"""
    sender: str
    receiver: str
    content: str
    timestamp: float

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

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

    def send(self, sender: str, receiver: str, content: str):
        message = Message(
            sender=sender,
            receiver=receiver,
            content=content,
            timestamp=time.time()
        )
        self.message_queue.append(message)

    def receive(self, agent_id: str) -> list[Message]:
        return [
            msg for msg in self.message_queue
            if msg.receiver == agent_id
        ]

 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

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

    def __init__(self):
        self.agents = {}
        self.protocol = CommunicationProtocol()

    def add_agent(self, agent_id: str, agent: Agent):
        self.agents[agent_id] = agent

    def orchestrate(self, task: Task) -> Result:
        """编排多个Agent协作完成任务"""

        # 1. 任务分解
        subtasks = self.decompose_task(task)

        # 2. 分配子任务
        assignments = self.assign_tasks(subtasks)

        # 3. 并行执行
        results = []
        for agent_id, subtask in assignments.items():
            agent = self.agents[agent_id]
            result = agent.execute(subtask)
            results.append(result)

            # 广播结果
            self.protocol.send(
                sender=agent_id,
                receiver="all",
                content=f"完成: {result}"
            )

        # 4. 整合结果
        return self.integrate_results(results)

 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

class SpecializedAgent(Agent):
    """专业化Agent"""

    def __init__(self, role: str, expertise: list[str]):
        self.role = role
        self.expertise = expertise
        super().__init__()

# 定义不同角色的Agent
researcher = SpecializedAgent(
    role="研究员",
    expertise=["信息检索", "数据分析", "报告撰写"]
)

coder = SpecializedAgent(
    role="程序员",
    expertise=["代码编写", "调试", "架构设计"]
)

reviewer = SpecializedAgent(
    role="审查员",
    expertise=["质量控制", "安全审计", "性能优化"]
)

# 协作流程
def collaborative_development(project: str):
    # 研究员：需求分析
    requirements = researcher.execute(
        f"分析项目需求: {project}"
    )

    # 程序员：实现代码
    code = coder.execute(
        f"根据需求实现: {requirements}"
    )

    # 审查员：代码审查
    review = reviewer.execute(
        f"审查代码: {code}"
    )

    return {
        "requirements": requirements,
        "code": code,
        "review": review
    }

 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

class CustomerServiceAgent(Agent):
    """智能客服Agent"""

    def __init__(self):
        super().__init__()

        # 工具集
        self.register_tool("search_knowledge_base", self.search_kb)
        self.register_tool("query_order_status", self.query_order)
        self.register_tool("create_ticket", self.create_ticket)
        self.register_tool("handover_to_human", self.handover)

        # 记忆系统
        self.memory = LongTermMemory(vector_db)
        self.working_memory = WorkingMemory(window_size=20)

    def handle_query(self, user_query: str, user_id: str):
        # 1. 检索用户历史
        history = self.memory.retrieve(
            query=f"用户{user_id}的历史对话",
            filter={"user_id": user_id}
        )

        # 2. 添加到工作记忆
        self.working_memory.add_message("user", user_query)

        # 3. 规划行动
        plan = self.planner.plan(
            query=user_query,
            context=history + self.working_memory.get_context()
        )

        # 4. 执行计划
        result = self.execute_plan(plan)

        # 5. 存储交互
        self.memory.store(
            content=f"用户: {user_query}\nAgent: {result}",
            metadata={
                "user_id": user_id,
                "timestamp": time.time(),
                "resolved": self.is_satisfied(result)
            }
        )

        return result

 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

class CodeAssistantAgent(Agent):
    """代码助手Agent"""

    def __init__(self):
        super().__init__()

        self.tools = {
            "read_file": self.read_file,
            "write_file": self.write_file,
            "execute_code": self.execute_code,
            "search_documentation": self.search_docs,
            "analyze_error": self.analyze_error,
            "refactor_code": self.refactor
        }

    def solve_problem(self, problem: str):
        # 1. 理解问题
        understanding = self.think(
            f"分析以下问题: {problem}"
        )

        # 2. 制定方案
        solution = self.plan(
            f"针对以下问题制定解决方案: {understanding}"
        )

        # 3. 实现代码
        code = self.generate_code(solution)

        # 4. 测试验证
        test_result = self.test(code)

        # 5. 如果失败，调试
        while not test_result.passed:
            debug_info = self.analyze_error(test_result.error)
            code = self.fix_code(code, debug_info)
            test_result = self.test(code)

        return code

 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

class ResearchAssistantAgent(Agent):
    """研究助手Agent"""

    def __init__(self):
        super().__init__()

        self.tools = {
            "search_paper": self.search_arxiv,
            "summarize_paper": self.summarize,
            "extract_citations": self.extract_citations,
            "find_related_work": self.find_related,
            "generate_latex": self.to_latex
        }

    def research_topic(self, topic: str) -> Report:
        # 1. 搜索相关论文
        papers = self.search_arxiv(topic, limit=20)

        # 2. 阅读和总结
        summaries = []
        for paper in papers:
            summary = self.summarize(paper)
            summaries.append(summary)

            # 存储记忆
            self.memory.store(
                content=summary,
                metadata={"paper_id": paper.id, "topic": topic}
            )

        # 3. 提取引用关系
        citations = self.extract_citations(papers)

        # 4. 查找相关工作
        related = self.find_related_work(citations)

        # 5. 生成报告
        report = self.generate_report(
            topic=topic,
            summaries=summaries,
            citations=citations,
            related_work=related
        )

        return report

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import asyncio

async def parallel_execution(agent: Agent, tasks: list[Task]):
    """并行执行多个任务"""
    coroutines = [
        asyncio.to_thread(agent.execute, task)
        for task in tasks
    ]

    results = await asyncio.gather(*coroutines)
    return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from functools import lru_cache

class CachedAgent(Agent):
    """带缓存的Agent"""

    @lru_cache(maxsize=1000)
    def think(self, query: str) -> Thought:
        """缓存思考结果"""
        return super().think(query)

    @lru_cache(maxsize=100)
    def plan(self, goal: str) -> Plan:
        """缓存规划结果"""
        return super().plan(goal)

1
2
3
4
5
6
7
8

async def stream_response(agent: Agent, query: str):
    """流式输出Agent响应"""
    async for chunk in agent.stream_execute(query):
        yield chunk

# 使用
async for chunk in stream_response(agent, "分析市场趋势"):
    print(chunk, end="", flush=True)

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

class SafeToolExecutor(ToolExecutor):
    """安全工具执行器"""

    def __init__(self):
        super().__init__()
        self.dangerous_tools = ["delete_file", "execute_system"]
        self.require_permission = []

    def execute(self, action: Action) -> Result:
        # 检查危险工具
        if action.tool_name in self.dangerous_tools:
            # 需要用户确认
            if not self.get_user_confirmation(action):
                return Result(error="用户取消操作")

        # 检查权限
        if action.tool_name in self.require_permission:
            if not self.check_permission(action):
                return Result(error="权限不足")

        return super().execute(action)

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

class ContentFilter:
    """内容过滤器"""

    def __init__(self):
        self.prohibited_patterns = [
            r"暴力.*内容",
            r"违法.*信息",
            # ...更多模式
        ]

    def filter(self, content: str) -> bool:
        """检查内容是否违规"""
        for pattern in self.prohibited_patterns:
            if re.search(pattern, content):
                return False
        return True

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

class PrivacyProtection:
    """隐私保护"""

    def __init__(self):
        self.sensitive_patterns = [
            r"\d{18}",  # 身份证
            r"\d{11}",  # 手机号
            r"\w+@\w+\.com",  # 邮箱
        ]

    def anonymize(self, text: str) -> str:
        """匿名化敏感信息"""
        for pattern in self.sensitive_patterns:
            text = re.sub(pattern, "***", text)
        return text

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

class AutonomousAgent(Agent):
    """完全自主的Agent"""

    def __init__(self):
        super().__init__()
        self.goal_system = GoalSystem()
        self.learning_system = LearningSystem()

    def set_goal(self, goal: str):
        """设置长期目标"""
        self.goal_system.set_goal(goal)

    def learn_from_experience(self):
        """从经验中学习"""
        experiences = self.memory.get_all()

        for exp in experiences:
            lesson = self.learning_system.learn(exp)
            self.memory.store(lesson, metadata={"type": "lesson"})

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

class MetaLearningAgent(Agent):
    """元学习Agent"""

    def __init__(self):
        super().__init__()
        self.task_library = TaskLibrary()

    def learn_to_learn(self):
        """学会学习"""
        # 从多个任务中提取共性
        tasks = self.task_library.get_all_tasks()

        meta_patterns = self.extract_patterns(tasks)

        # 应用到新任务
        self.apply_meta_patterns(meta_patterns)