fiuai-sdk-agent 0.4.3

FiuAI SDK Agent - A framework for building AI agents with LLM support, context management, and event handling

FiuAI SDK Agent

用于构建 AI Agent 的通用框架，提供运行时编排、Skill 注册、LLM 管理、向量检索等基础能力。

边界原则: SDK 只放抽象和可复用能力，不放业务语义。业务概念（如 Contract、Invoice）、业务字段映射、业务 Prompt 知识、业务判定策略等，全部由调用方注入或在业务项目中实现。

安装

# 基础安装
pip install fiuai-sdk-agent

# 带 Embedding 能力
pip install fiuai-sdk-agent[embedding]

# 带向量检索能力
pip install fiuai-sdk-agent[vector]

# 全部可选依赖
pip install fiuai-sdk-agent[all]

本仓库内开发时直接 editable 安装:

uv pip install -e ./fiuai_sdk_agent

架构分层

fiuai_sdk_agent/
├── core/             # 核心抽象 (AgentRuntime, SkillRegistry, PlanEngine, EventBus)
├── infra/            # 基础设施 (EmbeddingEngine, VectorStoreEngine)
├── graph/            # LangGraph 辅助 (BaseAgentState, create_node)
├── prompting/        # Prompt 模板引擎 (PromptTemplate)
├── agents/           # [Legacy] Agent 基类、类型定义、上下文管理
├── pkg/              # [Legacy] LLM 管理、HTTP 客户端
└── utils/            # 日志、异常定义

SDK 边界

属于 SDK	不属于 SDK
AgentRuntime 泛型运行时容器	业务对象定义 (Contract, Invoice...)
Skill 注册与执行框架	业务字段映射规则
PlanEngine / EventBus 抽象接口	业务 Prompt 知识
Embedding / Vector 抽象 + 通用实现	业务判定策略 (对账评分等)
LLM 管理、Callback、日志	业务特有的 Agent 实现
Prompt 模板引擎	业务 Prompt 内容

快速开始

1. 初始化 LLM 配置

from fiuai_sdk_agent.pkg.llm.llm_config import init_llm_config
from fiuai_sdk_agent.pkg.llm.types import LLMVendor

init_llm_config(configs=[
    LLMVendor(
        name="ali",
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        api_key="sk-xxx",
    ),
])

2. 使用 AgentRuntime 编排

from fiuai_sdk_agent import AgentRuntime, SkillRegistry
from fiuai_sdk_agent.pkg.llm import LLMManager

# 定义业务上下文 (由调用方定义, 不在 SDK 中)
@dataclass
class MyContext:
    tenant_id: str
    user_id: str

# 创建 Runtime
runtime: AgentRuntime[MyContext] = AgentRuntime(
    context=MyContext(tenant_id="t1", user_id="u1"),
    llm_manager=LLMManager(),
    skill_registry=SkillRegistry(),
)

# 获取 LLM 实例
llm = runtime.get_llm(model="qwen-plus")
response = llm.invoke("你好")

3. 注册 Skill

from pydantic import BaseModel
from fiuai_sdk_agent import SkillRegistry, SkillDefinition, AgentRuntime

class QueryInput(BaseModel):
    sql: str

class QueryOutput(BaseModel):
    rows: list

@SkillRegistry.register(
    name="run_query",
    description="执行 SQL 查询",
    input_model=QueryInput,
    output_model=QueryOutput,
    domain="data",
)
async def run_query(input_data: QueryInput, runtime: AgentRuntime) -> QueryOutput:
    # 业务逻辑
    return QueryOutput(rows=[])

# 导出为 OpenAI Tool Schema (供 LLM function calling)
tools = SkillRegistry.to_openai_tools(domain="data")

4. LangGraph Node

from fiuai_sdk_agent import BaseAgentState, create_node, AgentRuntime

async def my_node(state: BaseAgentState, runtime: AgentRuntime) -> dict:
    llm = runtime.get_llm()
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 包装为 LangGraph 节点
node = create_node(my_node, runtime)

5. Embedding + 向量检索

from fiuai_sdk_agent import OpenAICompatibleEmbeddingEngine, QdrantVectorStore

# Embedding (兼容 OpenAI / DashScope / Ollama 等)
embedding = OpenAICompatibleEmbeddingEngine(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="sk-xxx",
    model="text-embedding-v3",
    dimension=1024,
)
vectors = await embedding.embed_texts(["文本1", "文本2"])

# 向量存储 (Qdrant)
store = QdrantVectorStore(url="http://localhost:6333")
await store.ensure_collection("my_collection", dimension=1024)
await store.upsert("my_collection", id="doc1", vector=vectors[0], payload={"text": "文本1"})
results = await store.search("my_collection", vector=query_vector, limit=5)

6. Prompt 模板

from fiuai_sdk_agent import PromptTemplate

# 从字符串创建
tpl = PromptTemplate.from_string("你是{{role}}, 请回答: {{question}}")
prompt = tpl.render(role="财务专家", question="什么是增值税?")

# 从文件加载
tpl = PromptTemplate.from_file("prompts/my_prompt.txt")
prompt = tpl.render(strict=False, name="test")  # strict=False: 未提供的变量保留占位符

7. 实现 PlanEngine / EventBus (业务层)

SDK 只提供抽象接口, 业务层自行实现:

from fiuai_sdk_agent import PlanEngine, EventBus

class MyPlanManager(PlanEngine):
    def generate_plan(self, goal, context=None):
        ...
    def replan(self, plan, step_result):
        ...
    def start_step(self, plan, step_id):
        ...
    def complete_step(self, plan, step_id, result=None):
        ...
    def fail_step(self, plan, step_id, error):
        ...

class MyEventPublisher(EventBus):
    def publish(self, event):
        # 发送到 Redis / WebSocket / 消息队列等
        ...

核心 API

AgentRuntime[T]

泛型运行时上下文容器, T 为业务上下文类型。

方法	说明
get_llm(model, **kwargs)	获取 LLM 实例 (ChatOpenAI)
get_skill(name)	按名称获取 Skill
get_skills_by_domain(domain)	按领域获取 Skill 列表
get_skills_as_tools(domain)	导出为 OpenAI Tool Schema

SkillRegistry

Skill 注册中心, 支持装饰器注册。

方法	说明
register(name, description, input_model, output_model, domain, tags)	装饰器注册 Skill
get(name)	按名称获取 SkillDefinition
get_by_domain(domain)	按领域获取列表
to_openai_tools(domain)	导出为 OpenAI Tool Schema

EmbeddingEngine (抽象)

方法	说明
embed_text(text)	单条文本向量化
embed_texts(texts)	批量向量化
dimension	向量维度 (property)

内置实现: OpenAICompatibleEmbeddingEngine

VectorStoreEngine (抽象)

方法	说明
ensure_collection(name, dimension, distance)	确保 collection 存在
upsert(collection, id, vector, payload)	插入/更新
upsert_batch(collection, ids, vectors, payloads)	批量插入/更新
search(collection, vector, limit, score_threshold, filters)	向量搜索
delete(collection, ids)	按 ID 删除
collection_exists(name)	检查 collection 是否存在

内置实现: QdrantVectorStore

PlanEngine (抽象)

任务计划引擎, 业务层实现。

方法	说明
generate_plan(goal, context)	生成计划
replan(plan, step_result)	重新规划
start_step(plan, step_id)	开始步骤
complete_step(plan, step_id, result)	完成步骤
fail_step(plan, step_id, error)	标记失败

EventBus (抽象)

事件总线, 业务层实现。

方法	说明
publish(event)	发布事件

PromptTemplate

Prompt 模板引擎, 支持 {{variable}} 占位符。

方法	说明
from_string(template)	从字符串创建
from_file(path, encoding)	从文件加载
render(strict, **variables)	渲染模板
raw	原始模板文本 (property)

类型定义

SDK 提供完整的类型定义, 位于 fiuai_sdk_agent.agents.types:

• Agent 类型: AgentType, ExpertType
• Event 类型: EventType, Event, EventContext, EventData, ChunkData, TaskStatus
• Plan 类型: PlanStatus, PlanStepStatus, PlanStep, BasePlan, PlanStateMachine
• Decision 类型: DecisionType, DecisionOption, Decision, DecisionResult
• Artifact 类型: ArtifactType, Artifact

依赖

必需依赖

包	版本	用途
fiuai-sdk-python	>=0.6.5	FiuAI Python SDK
langchain	>=1.2.0	LLM 链式调用
langchain-openai	>=1.1.6	OpenAI 集成
langchain-core	>=1.2.0	LangChain 核心
pydantic	>=2.0.0	数据校验
pydantic-settings	>=2.0.0	配置管理
redis	>=7.1.0	缓存/事件队列
httpx	>=0.27.0	异步 HTTP
tenacity	>=8.0.0	重试
sqlalchemy	>=2.0.0	ORM
snowflake-id	>=1.0.2	ID 生成
jinja2	==3.1.6	模板引擎

可选依赖

Extra	包	用途
embedding	openai>=1.0.0	Embedding 引擎 (OpenAI 兼容 API)
vector	qdrant-client>=1.7.0	向量数据库 (Qdrant)
all	以上全部

Python 版本

• 要求 Python >= 3.12

发布

./publish_sdk_agent.sh

许可证

Copyright (c) 2025 FiuAI