xynehq/jaf-py
GitHub: xynehq/jaf-py
一个基于纯函数式编程范式的生产级 Python AI Agent 框架，提供不可变状态、MCP 支持、企业级安全防护与全链路可观测性。
Stars: 9 | Forks: 7
# JAF (Juspay Agent Framework) - Python 实现
[](https://github.com/xynehq/jaf-py)
[](https://www.python.org/)
[](https://xynehq.github.io/jaf-py/)
一个纯函数式的 agent 框架，具备不可变状态和可组合的 tools，由 TypeScript 专业转换至 Python。JAF 能够构建生产级 AI agent 系统，并内置安全性、可观测性和错误处理。
**生产就绪**：与 TypeScript 版本功能完全对齐，拥有全面的测试套件和生产部署支持。
## **[阅读完整文档](https://xynehq.github.io/jaf-py/)**
**[ 快速开始 →](https://xynehq.github.io/jaf-py/getting-started/)** | **[ API 参考 →](https://xynehq.github.io/jaf-py/api-reference/)** | **[ 示例 →](https://xynehq.github.io/jaf-py/examples/)**
## 核心特性
### **完整的 TypeScript 转换**
- **功能完全对齐**：所有 TypeScript 功能均已转换为 Python
- **类型安全**：带有运行时校验的 Pydantic 模型
- **不可变状态**：保留了函数式编程原则
- **Tool 集成**：完整的 tool 调用和执行系统
### **生产级就绪服务器**
- **FastAPI 服务器**：高性能异步 HTTP API
- **自动化文档**：通过 `/docs` 提供交互式 API 文档
- **健康监控**：内置健康检查和指标
- **CORS 支持**：为浏览器集成做好准备
### Model Context Protocol (MCP)
- **MCP Client**：完整的 MCP 规范支持
- **Stdio & SSE**：支持多种传输协议
- **Tool 集成**：无缝集成 MCP tool
- **自动发现**：从 MCP server 动态加载 tool
### **企业级安全**
- **输入 Guardrails**：内容过滤和校验
- **输出 Guardrails**：响应净化
- **权限系统**：基于角色的访问控制
- **审计日志**：完整的交互追踪
- **代理支持**：支持带身份验证的企业级代理集成
### **可观测性与监控**
- **实时追踪**：事件驱动的可观测性
- **OpenTelemetry 集成**：支持 OTLP 的分布式追踪
- **Langfuse 追踪**：LLM 可观测性与数据分析
- **结构化日志**：JSON 格式的日志
- **错误处理**：全面的错误类型和恢复机制
- **性能指标**：内置耗时统计与计数器
### **Agent-as-Tool 架构**
- **分层编排**：将 agent 作为其他 agent 中的 tool 使用
- **条件性 Tool 启用**：根据上下文启用/禁用 agent tool
- **会话管理**：为子 agent 配置会话继承
- **灵活的输出提取**：为 agent tool 输出提供自定义提取器
### **开发者体验**
- **CLI 工具**：项目初始化与管理
- **热重载**：具备自动重载功能的开发服务器
- **类型提示**：完全兼容 mypy
- **丰富的示例**：包含 RAG、多 agent、agent-as-tool 和 server 演示
- **可视化架构**：基于 Graphviz 生成的 agent 和 tool 关系图
## 核心理念
- **不可变性**：所有核心数据结构都是深度不可变的
- **纯函数**：核心逻辑以纯净、可预测的函数表达
- **边缘效应**：副作用被隔离在 Provider 模块中
- **组合优于配置**：通过组合简单的函数来构建复杂的行为
- **设计上的类型安全**：利用 Python 的类型系统和 Pydantic 保障运行时安全
## 快速开始
### 安装
```
# 从 GitHub 安装（开发版本）
pip install git+https://github.com/xynehq/jaf-py.git
# 或者安装所有可选依赖
pip install "jaf-py[all] @ git+https://github.com/xynehq/jaf-py.git"
# 安装特定功能集
pip install "jaf-py[server] @ git+https://github.com/xynehq/jaf-py.git" # FastAPI server support
pip install "jaf-py[memory] @ git+https://github.com/xynehq/jaf-py.git" # Redis/PostgreSQL memory providers
pip install "jaf-py[visualization] @ git+https://github.com/xynehq/jaf-py.git" # Graphviz visualization tools
pip install "jaf-py[tracing] @ git+https://github.com/xynehq/jaf-py.git" # OpenTelemetry and Langfuse tracing
pip install "jaf-py[dev] @ git+https://github.com/xynehq/jaf-py.git" # Development tools
```
### CLI 用法
JAF 包含一个用于项目管理的强大 CLI：
```
# 初始化新的 JAF 项目
jaf init my-agent-project
# 运行开发服务器
jaf server --host 0.0.0.0 --port 8000
# 显示版本和帮助
jaf version
jaf --help
```
### 开发环境设置
```
git clone https://github.com/xynehq/jaf-py
cd jaf-py
pip install -e ".[dev]"
# 运行测试
pytest
# 类型检查和 linting
mypy jaf/
ruff check jaf/
black jaf/
# 文档
pip install -r requirements-docs.txt
./docs.sh serve # Start documentation server
./docs.sh deploy # Deploy to GitHub Pages
```
## 文档
### **[官方文档网站](https://xynehq.github.io/jaf-py/)**
完整且支持搜索的文档可在 **[xynehq.github.io/jaf-py](https://xynehq.github.io/jaf-py/)** 获取，包含：
- **交互式导航**，支持搜索和过滤
- **暗黑/明亮模式**，支持自动检测系统偏好
- **移动端响应式设计**，适合在任何设备上查阅文档
- **实时代码示例**，带语法高亮
- **API 参考**，包含自动生成的文档
- **始终保持最新**，通过自动化部署同步
### **本地文档**
为了支持离线访问，文档也可以在 [`docs/`](docs/) 目录中找到：
- **[ 文档中心](docs/README.md)** - 您探索所有文档的起点
- **[ 快速入门](docs/getting-started.md)** - 安装说明与首个 agent 教程
- **[ 核心概念](docs/core-concepts.md)** - JAF 的函数式架构原则
- **[ API 参考](docs/api-reference.md)** - 完整的 Python API 文档
- **[ Tool 指南](docs/tools.md)** - 创建和使用 tool
- **[ 记忆系统](docs/memory-system.md)** - 持久化与记忆 provider
- **[ 模型 Provider](docs/model-providers.md)** - LiteLLM 集成
- **[ 监控](docs/monitoring.md)** - 可观测性、指标与预警
- **[ 服务器 API](docs/server-api.md)** - FastAPI endpoint 参考
- **[ 部署](docs/deployment.md)** - 生产环境部署指南
- **[ 示例](docs/examples.md)** - 详细的示例解析
- **[ 故障排除](docs/troubleshooting.md)** - 常见问题与解决方案
## 项目结构
```
jaf-py/
├── docs/ # Complete documentation
├── jaf/ # Core framework package
│ ├── core/ # Core types and engine
│ ├── memory/ # Conversation persistence
│ ├── providers/ # External integrations (LiteLLM, MCP)
│ ├── policies/ # Validation and security
│ ├── server/ # FastAPI server
│ └── cli.py # Command-line interface
├── examples/ # Example applications
└── tests/ # Test suite
```
## 架构可视化
JAF 包含强大的可视化功能，可帮助您理解和记录 agent 系统。
### 前置条件
首先，安装系统 Graphviz 依赖：
```
# macOS
brew install graphviz
# Ubuntu/Debian
sudo apt-get install graphviz
# Windows（通过 Chocolatey）
choco install graphviz
```
然后安装带有可视化支持的 JAF：
```
pip install "jaf-py[visualization]"
```
### 快速开始
```
import asyncio
from jaf import Agent, Tool, generate_agent_graph, GraphOptions
# 创建你的 agent
agent = Agent(
name='MyAgent',
instructions=lambda state: "I am a helpful assistant.",
tools=[my_tool],
handoffs=['OtherAgent']
)
# 生成可视化
async def main():
result = await generate_agent_graph(
[agent],
GraphOptions(
title="My Agent System",
output_path="./my-agents.png",
color_scheme="modern",
show_tool_details=True
)
)
if result.success:
print(f" Visualization saved to: {result.output_path}")
else:
print(f"❌ Error: {result.error}")
asyncio.run(main())
```
### 特性
- **多种配色方案**：可从 `default`、`modern` 或 `minimal` 主题中选择
- **Agent 架构**：可视化 agent、tool 及其移交关系
- **Tool 生态系统**：生成专用的 tool 交互图
- **Runner 架构**：展示包含会话层的完整系统架构
- **多种格式**：导出为 PNG、SVG 或 PDF
- **自定义布局**：支持多种 Graphviz 布局（`dot`、`circo`、`neato` 等）
### 输出示例
可视化系统会生成清晰、专业的图表，展示：
- **Agent 节点**：带有 agent 名称和模型信息的圆角矩形
- **Tool 节点**：展示 tool 名称和描述的椭圆
- **移交边**：指示 agent 移交关系的虚线
- **Tool 连接**：将 agent 连接到其 tool 的彩色边
- **集群组织**：在 runner 架构视图中对组件进行分组
### 高级用法
```
from jaf.visualization import run_visualization_examples
# 运行综合示例
await run_visualization_examples()
# 这将生成多个示例文件：
# - ./examples/agent-graph.png（agent 系统概览）
# - ./examples/tool-graph.png（tool 生态系统）
# - ./examples/runner-architecture.png（完整系统）
# - ./examples/agent-modern.png（现代配色方案）
```
## 核心组件
### 核心类型
```
from dataclasses import dataclass
from pydantic import BaseModel, Field
from jaf import Agent, Tool, RunState, run
# 定义你的 context 类型
@dataclass
class MyContext:
user_id: str
permissions: list[str]
# 定义 tool schema
class CalculateArgs(BaseModel):
expression: str = Field(description="Math expression to evaluate")
# 创建 tool
class CalculatorTool:
@property
def schema(self):
return type('ToolSchema', (), {
'name': 'calculate',
'description': 'Perform mathematical calculations',
'parameters': CalculateArgs
})()
async def execute(self, args: CalculateArgs, context: MyContext) -> str:
result = eval(args.expression) # Don't do this in production!
return f"{args.expression} = {result}"
# 定义 agent
def create_math_agent():
def instructions(state):
return 'You are a helpful math tutor'
return Agent(
name='MathTutor',
instructions=instructions,
tools=[CalculatorTool()]
)
```
### 运行框架
```
import asyncio
from jaf import run, make_litellm_provider, generate_run_id, generate_trace_id
from jaf.core.types import RunState, RunConfig, Message
async def main():
model_provider = make_litellm_provider('http://localhost:4000')
math_agent = create_math_agent()
config = RunConfig(
agent_registry={'MathTutor': math_agent},
model_provider=model_provider,
max_turns=10,
on_event=lambda event: print(event), # Real-time tracing
)
initial_state = RunState(
run_id=generate_run_id(),
trace_id=generate_trace_id(),
messages=[Message(role='user', content='What is 2 + 2?')],
current_agent_name='MathTutor',
context=MyContext(user_id='user123', permissions=['user']),
turn_count=0,
)
result = await run(initial_state, config)
print(result.outcome.output if result.outcome.status == 'completed' else result.outcome.error)
asyncio.run(main())
```
## 安全与校验
### 可组合的校验策略
```
from jaf.policies.validation import create_path_validator, create_permission_validator, compose_validations
# 创建独立验证器
path_validator = create_path_validator(['/shared', '/public'])
permission_validator = create_permission_validator('admin', lambda ctx: ctx)
# 组合它们
combined_validator = compose_validations(path_validator, permission_validator)
# 应用到 tool
secure_file_tool = with_validation(base_file_tool, combined_validator)
```
### Guardrails
```
from jaf.policies.validation import create_content_filter, create_length_guardrail
config = RunConfig(
# ... other config
initial_input_guardrails=[
create_content_filter(['spam', 'inappropriate']), # Requires blocked patterns
create_length_guardrail(max_length=1000, min_length=1)
],
final_output_guardrails=[
create_content_filter(['harmful', 'unsafe'])
],
)
```
## Agent-as-Tool 功能
JAF 2.2+ 引入了强大的 agent-as-tool 功能，允许您将 agent 作为 tool 在其他 agent 中使用，以实现分层编排：
```
from jaf.core.agent_tool import create_agent_tool
from jaf.core.types import create_json_output_extractor
# 创建专用 agent
spanish_agent = Agent(
name="spanish_translator",
instructions=lambda state: "Translate text to Spanish",
output_codec=TranslationOutput
)
french_agent = Agent(
name="french_translator",
instructions=lambda state: "Translate text to French",
output_codec=TranslationOutput
)
# 通过条件启用将 agent 转换为 tool
spanish_tool = spanish_agent.as_tool(
tool_name="translate_to_spanish",
tool_description="Translate text to Spanish",
max_turns=3,
custom_output_extractor=create_json_output_extractor(),
is_enabled=True # Always enabled
)
french_tool = french_agent.as_tool(
tool_name="translate_to_french",
tool_description="Translate text to French",
max_turns=3,
custom_output_extractor=create_json_output_extractor(),
is_enabled=lambda context, agent: context.language_preference == "french_spanish"
)
# 使用 agent tool 创建 orchestrator agent
orchestrator = Agent(
name="translation_orchestrator",
instructions=lambda state: "Use translation tools to respond in multiple languages",
tools=[spanish_tool, french_tool]
)
```
### 核心特性：
- **条件性启用**：根据运行时上下文启用/禁用 agent tool
- **会话管理**：配置子 agent 是否继承父会话状态
- **自定义输出提取**：定义如何提取和格式化 agent tool 的输出
- **错误处理**：针对失败的 agent tool 执行提供健壮的错误处理
## 🔗 Agent 移交
```
from jaf.policies.handoff import create_handoff_guardrail, HandoffPolicy
from pydantic import BaseModel
from enum import Enum
class AgentName(str, Enum):
MATH_TUTOR = 'MathTutor'
FILE_MANAGER = 'FileManager'
class HandoffOutput(BaseModel):
agent_name: AgentName
def create_triage_agent():
def instructions(state):
return 'Route requests to specialized agents'
return Agent(
name='TriageAgent',
instructions=instructions,
tools=[], # Regular tools would go here
handoffs=['MathTutor', 'FileManager'], # Allowed handoff targets
output_schema=HandoffOutput,
)
```
## 可观测性
### 实时追踪
```
from jaf.core.tracing import ConsoleTraceCollector, FileTraceCollector, create_composite_trace_collector
# 控制台日志
console_tracer = ConsoleTraceCollector()
# 文件日志
file_tracer = FileTraceCollector('./traces.log')
# 复合 tracing
tracer = create_composite_trace_collector(console_tracer, file_tracer)
config = RunConfig(
# ... other config
on_event=tracer.collect,
)
```
### OpenTelemetry 集成
JAF 2.2+ 内置了用于分布式追踪的 OpenTelemetry 支持：
```
import os
from jaf.core.tracing import create_composite_trace_collector, ConsoleTraceCollector
# 配置 OpenTelemetry endpoint
os.environ["TRACE_COLLECTOR_URL"] = "http://localhost:4318/v1/traces"
# 在创建 composite collector 时将自动配置 tracing
trace_collector = create_composite_trace_collector(ConsoleTraceCollector())
config = RunConfig(
# ... other config
on_event=trace_collector.collect,
)
```
### Langfuse 集成
用于针对 LLM 的专门可观测性和数据分析：
```
import os
from jaf.core.tracing import create_composite_trace_collector, ConsoleTraceCollector
# 配置 Langfuse 凭证
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-your-public-key"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-your-secret-key"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com" # or your self-hosted instance
# Langfuse tracing 将自动配置
trace_collector = create_composite_trace_collector(ConsoleTraceCollector())
config = RunConfig(
# ... other config
on_event=trace_collector.collect,
)
```
### 错误处理
```
from jaf.core.errors import JAFErrorHandler
if result.outcome.status == 'error':
formatted_error = JAFErrorHandler.format_error(result.outcome.error)
is_retryable = JAFErrorHandler.is_retryable(result.outcome.error)
severity = JAFErrorHandler.get_severity(result.outcome.error)
print(f"[{severity}] {formatted_error} (retryable: {is_retryable})")
```
## Provider 集成
### A2A (Agent-to-Agent) 通信
JAF 提供了一个健壮的 A2A 通信层，允许 agent 相互交互。这对于构建具有不同专业技能 agent 的多 agent 系统非常有用。
```
from jaf.a2a import create_a2a_agent, create_a2a_client, create_a2a_server
# 创建 agent
echo_agent = create_a2a_agent("EchoBot", "Echoes messages", "You are an echo bot.", [])
ping_agent = create_a2a_agent("PingBot", "Responds to pings", "You are a ping bot.", [])
# 创建 server
server_config = {
"agents": {"EchoBot": echo_agent, "PingBot": ping_agent},
"agentCard": {"name": "Test Server"},
"port": 8080,
}
server = create_a2a_server(server_config)
# 创建 client
client = create_a2a_client("http://localhost:8080")
```
### LiteLLM Provider
```
from jaf.providers.model import make_litellm_provider
# 连接到 LiteLLM proxy 以支持 100+ 模型
model_provider = make_litellm_provider(
'http://localhost:4000', # LiteLLM proxy URL
'your-api-key' # Optional API key
)
```
### MCP (Model Context Protocol) 集成
JAF 包含对 Model Context Protocol 的完整支持，以实现无缝的 tool 集成：
```
from jaf.providers.mcp import create_mcp_stdio_client, create_mcp_tools_from_client
# 通过 stdio 连接到 MCP server
mcp_client = create_mcp_stdio_client(['python', '-m', 'my_mcp_server'])
# 初始化并获取所有可用 tool
await mcp_client.initialize()
mcp_tools = await create_mcp_tools_from_client(mcp_client)
# 在你的 agent 中使用 MCP tool
def create_mcp_agent():
def instructions(state):
return "You have access to powerful MCP tools for various tasks."
return Agent(
name='MCPAgent',
instructions=instructions,
tools=mcp_tools # Automatically converted JAF tools
)
# 也支持 SSE client
from jaf.providers.mcp import create_mcp_sse_client
sse_client = create_mcp_sse_client('http://localhost:8080/sse')
```
## 开发服务器
JAF 包含一个内置的开发服务器，用于通过 HTTP endpoint 在本地测试 agent：
```
from jaf.server import run_server
from jaf.providers.model import make_litellm_provider
def create_my_agent():
def instructions(state):
return 'You are a helpful assistant'
return Agent(
name='MyAgent',
instructions=instructions,
tools=[calculator_tool, greeting_tool]
)
model_provider = make_litellm_provider('http://localhost:4000')
# 在端口 3000 上启动 server
await run_server(
[create_my_agent()],
{'model_provider': model_provider},
{'port': 3000}
)
```
服务器提供 RESTful endpoint：
- `GET /health` - 健康检查
- `GET /agents` - 列出可用 agent
- `POST /chat` - 通用聊天 endpoint
- `POST /agents/{name}/chat` - 指定 agent 的 endpoint
## 函数组合
JAF 支持函数式组合模式，可以通过简单、可重用的函数构建复杂的行为：
```
from jaf import create_function_tool, ToolSource
# 具有缓存和重试功能的增强型 tool
def with_cache(tool_func):
cache = {}
async def cached_execute(args, context):
cache_key = str(args)
if cache_key in cache:
return cache[cache_key]
result = await tool_func(args, context)
if result.status == "success":
cache[cache_key] = result
return result
return cached_execute
# 通过组合创建增强型 tool
enhanced_tool = create_function_tool({
'name': 'enhanced_search',
'description': 'Search with caching',
'execute': with_cache(base_search_function),
'parameters': SearchArgs,
'source': ToolSource.NATIVE
})
```
**核心优势：**
- **重用性**：一次编写，处处组合
- **可测试性**：每个函数都可以独立测试
- **类型安全**：支持完整的类型检查
- **性能**：可独立优化各个部分
## 示例应用
探索示例应用，观察框架的实际运行情况：
### 1. 多 Agent 服务器演示
```
cd examples
python server_example.py
```
**演示特性：**
- 多个专业 agent（数学、天气、通用）
- Tool 集成（计算器、天气 API）
- Agent 移交与路由
- 带有自动化文档的 RESTful API
- 实时追踪与错误处理
- 生产级服务器配置
**可用 endpoint：**
- `GET /health` - 服务器健康检查
- `GET /agents` - 列出所有可用 agent
- `POST /chat` - 与任何 agent 聊天
- `GET /docs` - 交互式 API 文档
### 2. Agent-as-Tool 演示
```
cd examples
python agent_as_tool_example.py
# 或作为 server 启动
python agent_as_tool_example.py --server
```
**演示特性：**
- 分层 agent 编排
- 基于上下文的条件性 tool 启用
- 从 agent tool 提取自定义输出
- 子 agent 的会话管理
- 翻译 agent 协同工作
### 3. 追踪集成演示
```
# OpenTelemetry tracing 示例
cd examples
python otel_tracing_demo.py
# Langfuse tracing 示例
python langfuse_tracing_demo.py
```
**演示特性：**
- OpenTelemetry 分布式追踪设置
- Langfuse LLM 可观测性集成
- 组合式 trace 收集器
- 实时监控与分析
### 4. MCP 集成演示
```
cd examples/mcp_demo
python main.py
```
**演示特性：**
- Model Context Protocol 集成
- 从 MCP server 动态加载 tool
- 安全的文件系统操作
- MCP client 配置与管理
## 测试
```
pytest # Run tests
ruff check . # Lint code
mypy . # Type checking
black . # Format code
```
## 架构原则
### 不可变状态机
- 所有状态转换都会创建新的状态对象
- 不修改现有的数据结构
- 状态转换可预测、可测试
### 类型安全
- 使用 Pydantic schema 进行运行时校验
- 使用 Python 类型提示提供编译时安全性
- 使用 NewType 实现类型安全的标识符
### 纯函数
- 核心逻辑无副作用
- 易于测试和推理
- 行为确定性
### 副作用隔离
- 副作用仅存在于 Provider 模块中
- 纯代码与非纯代码之间有清晰的边界
- 更容易进行模拟和测试
## 📜 许可证
MIT
## 🤝 贡献
1. Fork 该仓库
2. 创建一个功能分支
3. 为新功能添加测试
4. 运行测试套件
5. 提交 pull request
**JAF (Juspay Agentic Framework) v2.2** - 构建函数式 AI agent 系统的未来
标签：AI Agent框架, AV绕过, FastAPI, MCP, Python, 人工智能, 函数式编程, 搜索引擎查询, 无后门, 测试用例, 用户代理, 用户模式Hook绕过, 逆向工具