AI Agent 开发实战：从零构建你的第一个智能代理（附完整代码）

1次阅读

共计 2378 个字符，预计需要花费 6 分钟才能阅读完成。

AI Agent（智能代理）的核心可以简化为 PEAS 模型：感知 (Perception)、决策 (Decision)、执行 (Action)、学习 (Study)。这就像人类处理问题的过程：先收集信息，然后思考怎么做，最后采取行动，并从结果中学习。

传统 Rule-based 架构 ：依赖预先编写的规则，比如 ” 如果用户说 ’ 你好 ’，就回复 ’ 你好呀 '”。优点是响应快、可控性强，缺点是灵活性差，需要人工维护大量规则。
LLM-based 架构 ：利用大语言模型（如 GPT）的通用能力。优点是能处理开放性问题，缺点是响应速度慢、成本高，且需要精心设计提示词（Prompt）。

实际开发中，我们通常采用混合架构：用 LLM 处理复杂语义，用规则系统保障关键流程。

异步响应处理不当 ：直接同步调用 LLM 会导致界面卡死。正确的做法是使用 async/await 机制，就像点外卖后先去干别的，等餐到了再处理。
遗忘对话历史 ：每次交互都把上下文重新发给 LLM。这就像每次聊天都失忆，不仅效率低，API 调用成本也飙升。
过度依赖单一 LLM：把所有逻辑都塞进 Prompt。当 LLM 服务不稳定时，整个系统就会崩溃，应该像投资一样做技术组合。

我们用一个 Python 类来实现基础 Agent 框架，主要包含这些部件：

class BaseAgent:
    def __init__(self):
        self.memory = []  # 对话记忆
        self.state = "IDLE"  # 状态机

    async def perceive(self, input_text):
        """感知输入并更新状态"""
        self.memory.append(f"User: {input_text}")

    async def decide(self):
        """决策引擎核心"""
        if "紧急" in self.memory[-1]:
            self.state = "URGENT"
        # 其他状态转换逻辑...

    async def execute(self):
        """执行动作"""
        if self.state == "URGENT":
            return "已启动应急预案"
        # 对接 LLM 的异步调用
        response = await call_llm("
".join(self.memory[-3:]))
        return response

与 LangChain 集成的关键技巧：

用 ConversationChain 管理多轮对话
通过 Memory 类实现自动上下文修剪
用 LCEL（LangChain Expression Language）组合不同组件

下面是一个具备记忆功能的天气查询 Agent：

import asyncio
from typing import List, Dict
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

class WeatherAgent:
    """带异常处理的异步 Agent 示例"""

    def __init__(self):
        self.memory = ConversationBufferMemory()
        self.llm = ChatOpenAI(temperature=0.5)
        self.chain = ConversationChain(
            llm=self.llm,
            memory=self.memory
        )

    async def respond(self, query: str) -> str:
        """处理用户输入"""
        try:
            # 异步调用避免阻塞
            result = await asyncio.wait_for(self.chain.arun(input=query),
                timeout=10.0
            )

            # 敏感信息过滤
            if "密码" in result:
                result = "涉及隐私内容已屏蔽"

            return result

        except asyncio.TimeoutError:
            return "响应超时，请稍后再试"
        except Exception as e:
            print(f"Error: {e}")
            return "服务暂时不可用"

# 使用示例
async def main():
    agent = WeatherAgent()
    print(await agent.respond("北京今天天气怎么样？"))
    print(await agent.respond("那上海呢？"))  # 能记住上一句的上下文

asyncio.run(main())