Agent As MCP

乌鸦技能

一个使用基于技能的方法构建自适应AI代理的库。代理会自动为任务选择最佳技能，或从成功的对话中生成新技能。

与标准RAG或代理系统不同， raven-skills 实现 通过经验学习：代理记住成功的解决方案，并将其转化为可重用的技能。

主要特点

• 🎯 自动技能匹配 --通过嵌入相似性找到最合适的技能
• 🔧 技能生成 --从成功的对话中创造新技能
• 🧠 内置LLM客户端 --包括所有提示逻辑和SGR（模式引导推理）
• 🔄 优化 --相似技能的自动合并
• ⚡ 适应性改进 --修复执行失败的三种策略
• 💬 对话代理 --带有澄清和工具支持的多回合对话

安装

pip install raven-skills

快速开始

图书馆的设计很简单。你只需要一个OpenAI客户端和一个技能存储。

import asyncio
from openai import AsyncOpenAI
from raven_skills import SkillDialogueAgent, Tool, JSONStorage

# Define a tool (optional)
def get_weather(city: str) -> str:
    return f"Weather in {city}: sunny, 22°C"

weather_tool = Tool(
    name="get_weather",
    description="Get weather forecast for a city",
    parameters={"type": "object", "properties": {"city": {"type": "string"}}},
    function=get_weather,
)

async def main():
    # 1. Initialize
    agent = SkillDialogueAgent(
        client=AsyncOpenAI(),
        storage=JSONStorage("./skills.json"),
        tools=[weather_tool],
        auto_generate_skills=True,
    )

    # 2. Chat with the agent
    response = await agent.chat("What's the weather in Moscow?")
    print(response.message)
    # → "It's sunny in Moscow, 22°C"
    # Agent created skill "Get Weather Forecast" and saved it
    
    # 3. If something goes wrong, provide feedback
    response = await agent.chat("No, that's wrong - I asked about Paris")
    print(response.message)
    # → Agent detects negative feedback, refines skill, retries
    
    # 4. Save refined skill if needed
    if response.skill_refined:
        await agent.save_refined_skill()

if __name__ == "__main__":
    asyncio.run(main())

运作原理

SkillAgent 是唯一的切入点。在内部，它封装了所有LLM复杂性：

1. 提示和模板：所有系统提示都是预先编写和优化的。
2. 模式引导推理（SGR）代理不仅生成文本，还填充严格的Pydantic模式。这确保了技能步骤始终采用正确的格式。
3. 嵌入：通过OpenAI客户端自动生成搜索向量。

无需编写自己的提示或解析器。

技能存储

技能存储使用 SkillStorage 界面。图书馆包括 JSONStorage 用于基于文件的持久性。对于生产环境，您可以实现自己的（例如，Postgres+pgvector）。

定制存储实施

from raven_skills import SkillStorage, Skill

class PostgresStorage(SkillStorage):
    def __init__(self, db_pool):
        self.db = db_pool

    async def save(self, skill: Skill) -> None:
        # Your SQL INSERT ...
        pass

    async def get(self, skill_id: str) -> Skill | None:
        # Your SQL SELECT ...
        pass
        
    async def get_all(self) -> list[Skill]:
        # Get all skills
        pass

    async def delete(self, skill_id: str) -> None:
        # Delete skill
        pass

    async def search_by_embedding(
        self, embedding: list[float], top_k: int = 5, min_score: float = 0.0
    ) -> list[tuple[Skill, float]]:
        # Vector similarity search
        pass

高级功能

诊断和自我纠正

使用SkillDialogueAgent （推荐）：只需在对话中提供负面反馈：

# The agent automatically diagnoses and refines
response = await agent.chat("That's wrong, step 2 has an auth error")
# Agent: diagnoses → refines skill → retries automatically

if response.skill_refined:
    await agent.save_refined_skill()

使用SkillAgent （低级）：手动诊断和改进：

from raven_skills import SkillAgent

agent = SkillAgent(client=AsyncOpenAI(), storage=storage)

# Execute skill
execution = await agent.execute(skill, task)

# Diagnose issue
action = await agent.diagnose(
    skill=skill,
    task=task,
    result=execution,
    user_feedback="Step 2 threw an auth error",
)

print(f"Diagnosis: {action.diagnosis}")
# e.g.: "wrong_steps" → "Error in skill steps"

# Refine (creates new version of skill)
refined_skill = await agent.refine(skill, action)

合并与优化

随着时间的推移，重复的技能可能会出现。使用 SkillAgent 对于优化任务：

from raven_skills import SkillAgent, JSONStorage
from openai import AsyncOpenAI

# SkillAgent provides low-level skill operations including optimization
agent = SkillAgent(
    client=AsyncOpenAI(),
    storage=JSONStorage("./skills.json"),
)

# Find similar skills and merge them
results = await agent.optimize(similarity_threshold=0.95, dry_run=False)

for originals, merged in results:
    print(f"Merged {len(originals)} skills into '{merged.name}'")

数据模型

技能

@dataclass
class Skill:
    id: str
    name: str                  # Name
    metadata: SkillMetadata    # description, goal, keywords, embedding
    steps: list[SkillStep]     # Execution steps
    version: int               # Skill version
    parent_id: str | None      # Parent ID (if forked)

任务

@dataclass
class Task:
    id: str
    query: str                # Original user query
    key_aspects: list[str]    # Key aspects extracted by LLM
    embedding: list[float]    # Query vector

对话代理

对于包含澄清问题的交互式场景，请使用 SkillDialogueAgent:

from raven_skills import SkillDialogueAgent, Tool

# Define tools
weather_tool = Tool(
    name="get_weather",
    description="Get weather forecast",
    parameters={"type": "object", "properties": {"city": {"type": "string"}}},
    function=lambda city: f"Weather in {city}: +10°C",
)

agent = SkillDialogueAgent(
    client=AsyncOpenAI(),
    storage=storage,
    tools=[weather_tool],
    auto_generate_skills=True,  # Generate skills automatically
)

# Dialogue with clarifications
response = await agent.chat("What's the weather?")
# Agent: "Which city?" (needs_user_input=True)

response = await agent.chat("Moscow")
# Agent: "Weather in Moscow: +10°C" (called get_weather)

# The "weather" skill is now saved and will be reused

反馈和错误纠正

当代理出错时，提供反馈：

# Automatic detection: just tell what's wrong
response = await agent.chat("No, you missed the second sheet")
# Agent detects negative feedback → refines skill → retries

# Or use explicit API
response = await agent.feedback("negative", "You missed the second sheet")
# → Agent diagnoses error, refines skill, and retries

# Save improved skill after successful retry
if response.skill_refined:
    await agent.save_refined_skill()

# Positive feedback
await agent.feedback("positive")

LangChain/LangGraph集成

该库与LangChain生态系统完全兼容。

LangChain工具

from raven_skills.integrations import SkillMatcherTool, SkillDialogueTool

# As a tool in a LangChain agent
skill_tool = SkillMatcherTool(agent=skill_agent)
dialogue_tool = SkillDialogueTool(agent=dialogue_agent)

# Use in AgentExecutor or chains
from langchain.agents import AgentExecutor
agent_executor = AgentExecutor(agent=..., tools=[skill_tool])

LangChain可运行（LCEL）

from raven_skills.integrations import SkillAgentRunnable

# As a Runnable in LCEL chain
skill_runnable = SkillAgentRunnable(agent=dialogue_agent)

chain = prompt | skill_runnable | output_parser
result = await chain.ainvoke({"query": "Book a restaurant"})

LangGraph状态机

from langgraph.graph import StateGraph
from raven_skills.integrations import (
    create_skill_node,
    create_skill_router,
    SkillGraphState,
)

# Create state graph
graph = StateGraph(SkillGraphState)

# Add node with our agent
skill_node = create_skill_node(dialogue_agent)
graph.add_node("skill_agent", skill_node)

# Routing: if user input needed — wait, otherwise complete
router = create_skill_router(
    needs_input_node="wait_for_input",
    complete_node="__end__",
)
graph.add_conditional_edges("skill_agent", router)

# Compile and run
app = graph.compile()
result = await app.ainvoke({"current_message": "Order pizza"})

可用适配器

MCP 服务器

将raven技能作为MCP服务器运行，以便与Claude、Cursor和其他MCP客户端一起使用。

安装

pip install raven-skills[mcp]

快速开始

# Default server with OpenAI
python -m raven_skills serve

# With Ollama
python -m raven_skills serve --base-url http://localhost:11434/v1 --model llama3

所有CLI选项

python -m raven_skills serve [APP] [OPTIONS]

# APP (optional): Custom app module:variable (e.g., my_app:mcp)

Options:
  --port PORT              HTTP port (default: 8000)
  --transport {http,stdio} Transport type (default: http)
  --storage PATH           Skills JSON file (default: ./skills.json)
  --model MODEL            LLM model (default: gpt-4o-mini)
  --embedding-model MODEL  Embedding model (default: text-embedding-3-small)
  --base-url URL           OpenAI-compatible API URL
  --api-key KEY            API key (overrides OPENAI_API_KEY)
  --embedding-base-url URL Separate URL for embeddings

自定义应用程序（FastAPI风格）

在Python中创建完全配置的服务器，使用CLI运行：

# my_server.py
from openai import AsyncOpenAI
from mcp.server.fastmcp import FastMCP
from raven_skills import SkillDialogueAgent, Tool, JSONStorage

# 1. Configure your tools
def get_weather(city: str) -> str:
    return f"Weather in {city}: sunny, 22°C"

def search_restaurants(location: str, cuisine: str = "any") -> str:
    return f"Found 5 {cuisine} restaurants near {location}"

tools = [
    Tool(
        name="get_weather",
        description="Get current weather",
        parameters={"type": "object", "properties": {"city": {"type": "string"}}},
        function=get_weather,
    ),
    Tool(
        name="search_restaurants",
        description="Find restaurants",
        parameters={"type": "object", "properties": {
            "location": {"type": "string"},
            "cuisine": {"type": "string"},
        }},
        function=search_restaurants,
    ),
]

# 2. Configure client and storage
client = AsyncOpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
storage = JSONStorage("./my-skills.json")

# 3. Create agent with everything configured
agent = SkillDialogueAgent(
    client=client,
    storage=storage,
    tools=tools,
    llm_model="llama3",
    embedding_model="bge-m3:latest",
    auto_generate_skills=True,
)

# 4. Create MCP server and expose agent
mcp = FastMCP(name="my-skills-server")

@mcp.tool()
async def chat(message: str) -> str:
    """Chat with the skill agent."""
    response = await agent.chat(message)
    return response.message

@mcp.tool()
async def list_skills() -> list[dict]:
    """List all learned skills."""
    skills = await storage.get_all()
    return [{"name": s.name, "description": s.metadata.description} for s in skills]

@mcp.tool()
async def reset_conversation() -> str:
    """Reset the conversation state."""
    agent.reset()
    return "Conversation reset"

运行它：

python -m raven_skills serve my_server:mcp --port 9000

可用工具

可用资源

Claude桌面集成

添加到您的Claude桌面配置(~/Library/Application Support/Claude/claude_desktop_config.json):

{
  "mcpServers": {
    "raven-skills": {
      "command": "python",
      "args": ["-m", "raven_skills", "serve", "--transport", "stdio"]
    }
  }
}

许可证

Apache 2.0