AI Agent入门实战：从零构建智能任务代理系统

1次阅读

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

AI Agent 正在重塑人机交互方式：在电商客服场景可 7 ×24 小时处理 80% 的常规咨询（如订单查询 / 退换货），智能运维领域能自动诊断服务器故障并执行修复脚本，个人助理场景则可协调多个 API 完成复杂任务（如 ” 帮我订明天下午会议室并邀请张总 ”）。其核心价值在于将大语言模型 (LLM) 的认知能力与实际业务流程相连接。

优势：
内置工具集成 (Toolkits) 支持 170+ 服务
提供记忆 (Memory) 和会话管理开箱即用
活跃社区持续更新
劣势：
抽象层级较高导致响应延迟增加 200-300ms
定制化工作流需绕开框架约束

优势：
可针对性优化关键路径（如函数调用延迟 <50ms）
架构完全适配业务需求
无第三方依赖风险
劣势：
需自行实现状态管理等基础组件
工具扩展需重复造轮子

决策建议：短期验证用 LangChain，性能敏感场景建议自主开发。

# 工具注册示例（带类型提示）from typing import Literal

def get_weather(
    location: str, 
    unit: Literal['celsius', 'fahrenheit'] = 'celsius'
) -> str:
    """获取指定城市天气数据"""
    # 实际调用气象 API 的代码
    return f"{location}当前 25{unit[:1]}"

# 工具描述必须严格遵循 OpenAI 格式
tools = [{
    "name": "get_weather",
    "description": "查询城市天气",
    "parameters": {
        "type": "object",
        "properties": {"location": {"type": "string"},
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
        },
        "required": ["location"]
    }
}]

from collections import deque
from dataclasses import dataclass

@dataclass
class DialogueState:
    history: deque = deque(maxlen=10)  # 限制记忆窗口
    pending_tools: dict = None        # 等待回调的工具
    user_context: dict = None         # 用户画像数据

def handle_message(state: DialogueState, user_input: str) -> str:
    try:
        # 调用 LLM 生成响应或工具调用
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[dict(role="user", content=user_input)],
            tools=tools
        )

        # 处理工具调用情形
        if tool_calls := response.choices[0].message.tool_calls:
            state.pending_tools = {"call_id": tool_calls[0].id,
                "args": json.loads(tool_calls[0].function.arguments)
            }
            return "正在处理您的请求..."

        # 更新对话历史
        state.history.append((user_input, response.choices[0].message.content))
        return response.choices[0].message.content

    except Exception as e:
        # 异常恢复逻辑
        state.pending_tools = None
        return f"系统错误: {str(e)}"

from pydantic import BaseModel, Field, validator

class UserQuery(BaseModel):
    text: str = Field(min_length=1, max_length=500)
    user_id: int = Field(gt=0)

    @validator('text')
    def filter_injection(cls, v):
        if 'DROP TABLE' in v.upper():
            raise ValueError("检测到恶意输入")
        return v

import asyncio

async def parallel_tool_execution(tasks: list):
    """并发执行多个工具调用"""
    return await asyncio.gather(*[asyncio.to_thread(tool.func, **tool.args) 
          for tool in tasks]
    )

def compress_history(history: list) -> list:
    """基于 TF-IDF 保留关键对话片段"""
    from sklearn.feature_extraction.text import TfidfVectorizer

    texts = [f"Q: {q}\nA: {a}" for q, a in history]
    vectorizer = TfidfVectorizer(max_features=5)
    X = vectorizer.fit_transform(texts)

    # 取权重最高的 3 轮对话
    return [history[i] for i in X.sum(axis=1).argsort()[-3:]]

令牌桶算法实现请求排队
失败请求的指数退避重试
本地缓存高频查询结果

输入阶段：正则匹配手机 / 身份证号

输出阶段：LLM 提示词中加入

你不得透露以下信息：- 用户真实联系方式
- 系统内部 IP 地址

MAX_TURNS = 5
def detect_loop(state: DialogueState) -> bool:
    recent = [msg[1] for msg in list(state.history)[-MAX_TURNS:]]
    return len(set(recent)) < MAX_TURNS/2