科大讯飞AstronClaw发布后:Python实战从零构建AI Agent系统

import json
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Message:
    role: str  # "user" / "assistant" / "system"
    content: str
    timestamp: float = field(default_factory=time.time)

@dataclass
class Tool:
    name: str
    description: str
    input_schema: Dict[str, Any]
    function: callable  # 实际执行函数

class Memory:
    """记忆模块：管理对话历史和任务上下文"""

    def __init__(self, max_history: int = 100):
        self.messages: List[Message] = []
        self.max_history = max_history
        self.context: Dict[str, Any] = {}  # 跨对话的持久上下文
        self.task_history: List[Dict] = []  # 历史任务记录

    def add_message(self, role: str, content: str):
        self.messages.append(Message(role=role, content=content))
        # 防止内存溢出，只保留最近N条
        if len(self.messages) > self.max_history:
            self.messages = self.messages[-self.max_history:]

    def get_context_window(self, window_size: int = 20) -> List[Message]:
        """获取最近N条消息，用于送入大模型"""
        return self.messages[-window_size:]

    def store_fact(self, key: str, value: Any):
        self.context[key] = value

    def get_fact(self, key: str, default: Any = None) -> Any:
        return self.context.get(key, default)

class ToolRegistry:
    """工具注册中心：管理所有可用工具"""

    def __init__(self):
        self.tools: Dict[str, Tool] = {}

    def register(self, name: str, description: str,
                input_schema: Dict[str, Any],
                function: callable):
        self.tools[name] = Tool(
            name=name,
            description=description,
            input_schema=input_schema,
            function=function
        )

    def execute(self, tool_name: str, **kwargs) -> Any:
        if tool_name not in self.tools:
            raise ValueError(f"Tool {tool_name} not found")
        tool = self.tools[tool_name]
        try:
            result = tool.function(**kwargs)
            return result
        except Exception as e:
            return {"error": str(e)}

    def list_tools(self) -> List[Dict]:
        return [
            {"name": t.name, "description": t.description,
              "input_schema": t.input_schema}
            for t in self.tools.values()
        ]

# 工具1：获取股票实时行情
def get_stock_quote(stock_code: str) -> Dict:
    """获取股票实时报价（示例用akshare）"""
    import akshare as ak
    try:
        # 获取单只股票实时数据
        df = ak.stock_zh_a_spot_em()
        row = df[df['代码'] == stock_code]
        if row.empty:
            return {"error": f"Stock {stock_code} not found"}
        data = row.iloc[0].to_dict()
        return {
            "code": stock_code,
            "name": data.get("名称", ""),
            "price": data.get("最新价", 0),
            "change_pct": data.get("涨跌幅", 0),
            "volume": data.get("成交量", 0),
            "turnover": data.get("成交额", 0)
        }
    except Exception as e:
        return {"error": str(e)}

# 工具2：获取股票历史K线数据
def get_stock_history(stock_code: str, period: str = "daily",
                           adjust: str = "qfq") -> Dict:
    """获取股票历史K线数据"""
    import akshare as ak
    try:
        df = ak.stock_zh_a_hist(
            symbol=stock_code,
            period=period,
            adjust=adjust,
            start_date="20260101",
            end_date="20260416",
            limit=30  # 最近30个交易日
        )
        # 计算简单技术指标
        df["MA5"] = df["收盘"].rolling(5).mean()
        df["MA20"] = df["收盘"].rolling(20).mean()
        return {
            "code": stock_code,
            "latest_price": df["收盘"].iloc[-1],
            "MA5": round(df["MA5"].iloc[-1], 2),
            "MA20": round(df["MA20"].iloc[-1], 2),
            "volume_ratio": round(
                df["成交量"].iloc[-1] / df["成交量"].rolling(20).mean().iloc[-1], 2
            )
        }
    except Exception as e:
        return {"error": str(e)}

class StockAnalysisAgent:
    """股票分析Agent：整合感知、规划、行动、记忆"""

    def __init__(self, llm_client, tools: ToolRegistry):
        self.llm = llm_client  # 接入任意LLM API
        self.tools = tools
        self.memory = Memory()

    def think(self, user_input: str) -> str:
        """Agent核心：思考-规划-执行循环"""
        # Step 1: 将用户输入存入记忆
        self.memory.add_message("user", user_input)

        # Step 2: 构建系统提示词（含工具描述）
        tool_descs = "\n".join([
            f"- {t['name']}: {t['description']}"
            for t in self.tools.list_tools()
        ])
        system_prompt = f"""
    你是一个股票分析助手。
    用户提出股票相关问题，你可以调用以下工具：
    {tool_descs}
    每次只调用一个工具。如果需要多个工具，按顺序逐一调用。
    """

        # Step 3: 与LLM对话直到任务完成
        MAX_ITERATIONS = 5
        for _ in range(MAX_ITERATIONS):
            # 构建消息历史
            messages = [
                {"role": "system", "content": system_prompt},
                *[{"role": m.role, "content": m.content}
                  for m in self.memory.get_context_window()]
            ]
            response = self.llm.chat(messages)

            # 解析LLM响应：判断是最终回答还是调用工具
            if response.is_final_answer:
                self.memory.add_message("assistant", response.content)
                return response.content
            else:
                # 执行工具调用
                tool_name = response.tool_to_call
                tool_result = self.tools.execute(**response.tool_args)
                self.memory.add_message("assistant",
                     f"[调用工具: {tool_name}]\n结果: {tool_result}")

        return "任务超时，未能在规定步数内完成。"

# 初始化Agent
tools = ToolRegistry()
tools.register(
    "get_quote",
    "获取股票实时报价数据，包括最新价、涨跌幅、成交量",
    {"stock_code": "str"},
    get_stock_quote
)
tools.register(
    "get_history",
    "获取股票历史K线数据，计算均线和量比指标",
    {"stock_code": "str", "period": "str"},
    get_stock_history
)

# 假设有一个LLM客户端（支持OpenAI兼容API）
llm = OpenAICompatibleClient(api_key="your-key", base_url="...")
agent = StockAnalysisAgent(llm, tools)

# 运行分析
user_question = "帮我分析一下科大讯飞的股票，值得关注吗？"
answer = agent.think(user_question)
print(answer)