microsoft-agent-framework by rysweet/amplihack
npx skills add https://github.com/rysweet/amplihack --skill microsoft-agent-framework版本 : 0.1.0-preview | 最后更新 : 2025-11-15 | 框架版本 : 0.1.0-preview 语言 : Python 3.10+, C# (.NET 8.0+) | 许可证 : MIT
Microsoft Agent Framework 是一个用于构建生产级 AI 智能体和工作流的开源平台,它融合了 AutoGen 的简洁性和 Semantic Kernel 的企业级特性。
核心能力 : AI 智能体(有状态对话、工具集成) | 工作流(基于图的编排、并行处理) | 企业级特性(遥测、中间件、MCP 支持)
安装 :
pip install agent-framework-core --predotnet add package Microsoft.Agents.AI --prerelease仓库 : https://github.com/microsoft/agent-framework (5.1k stars)
在以下场景中使用 Microsoft Agent Framework:
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
与 amplihack 集成 : 使用 Agent Framework 处理有状态对话智能体和复杂编排。使用 amplihack 的原生智能体系统处理无状态任务委派和简单编排。详细指南请参阅 @integration/decision-framework.md。
能够处理消息、调用工具并维护上下文的有状态对话实体。
Python 示例 :
from agents_framework import Agent, ModelClient
# 使用模型创建智能体
agent = Agent(
name="assistant",
model=ModelClient(model="gpt-4"),
instructions="You are a helpful assistant"
)
# 单轮对话
response = await agent.run(message="Hello!")
print(response.content)
# 多轮对话(使用线程)
from agents_framework import Thread
thread = Thread()
response = await agent.run(thread=thread, message="What's 2+2?")
response = await agent.run(thread=thread, message="Double that")
C# 示例 :
using Microsoft.Agents.AI;
var agent = new Agent(
name: "assistant",
model: new ModelClient(model: "gpt-4"),
instructions: "You are a helpful assistant"
);
var response = await agent.RunAsync("Hello!");
Console.WriteLine(response.Content);
通过提供可调用函数来扩展智能体能力。
Python 示例 :
from agents_framework import function_tool
@function_tool
def get_weather(location: str) -> str:
"""Get weather for a location."""
return f"Weather in {location}: Sunny, 72°F"
agent = Agent(
name="assistant",
model=ModelClient(model="gpt-4"),
tools=[get_weather]
)
response = await agent.run(message="What's the weather in Seattle?")
# 智能体自动调用 get_weather() 并返回结果
C# 示例 :
[FunctionTool]
public static string GetWeather(string location)
{
return $"Weather in {location}: Sunny, 72°F";
}
var agent = new Agent(
name: "assistant",
model: new ModelClient(model="gpt-4"),
tools: new[] { typeof(Tools).GetMethod("GetWeather") }
);
用于多智能体系统的基于图的编排,支持条件路由和并行执行。
Python 示例 :
from agents_framework import Workflow, GraphWorkflow
# 定义工作流图
workflow = GraphWorkflow()
# 添加智能体作为节点
workflow.add_node("researcher", research_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("reviewer", review_agent)
# 定义边(控制流)
workflow.add_edge("researcher", "writer") # 顺序执行
workflow.add_edge("writer", "reviewer")
# 条件路由
def should_revise(state):
return state.get("needs_revision", False)
workflow.add_conditional_edge(
"reviewer",
should_revise,
{"revise": "writer", "done": "END"}
)
# 执行工作流
result = await workflow.run(initial_message="Research AI trends")
C# 示例 :
var workflow = new GraphWorkflow();
workflow.AddNode("researcher", researchAgent);
workflow.AddNode("writer", writerAgent);
workflow.AddNode("reviewer", reviewAgent);
workflow.AddEdge("researcher", "writer");
workflow.AddEdge("writer", "reviewer");
var result = await workflow.RunAsync("Research AI trends");
跨智能体维护对话历史和共享状态。
Python :
from agents_framework import Thread, ContextProvider
# Thread 维护对话历史
thread = Thread()
await agent.run(thread=thread, message="Remember: My name is Alice")
await agent.run(thread=thread, message="What's my name?") # "Alice"
# 自定义上下文提供者
class DatabaseContext(ContextProvider):
async def get_context(self, thread_id: str):
return await db.fetch_history(thread_id)
async def save_context(self, thread_id: str, messages):
await db.save_history(thread_id, messages)
agent = Agent(model=model, context_provider=DatabaseContext())
添加日志记录、身份验证和监控等横切关注点。
Python :
from agents_framework import Middleware
from opentelemetry import trace
# 自定义中间件
class LoggingMiddleware(Middleware):
async def process(self, message, next_handler):
print(f"Processing: {message.content}")
response = await next_handler(message)
print(f"Response: {response.content}")
return response
# OpenTelemetry 集成
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("agent-run"):
response = await agent.run(message="Hello")
C# :
public class LoggingMiddleware : IMiddleware
{
public async Task<Message> ProcessAsync(Message message, Func<Message, Task<Message>> next)
{
Console.WriteLine($"Processing: {message.Content}");
var response = await next(message);
Console.WriteLine($"Response: {response.Content}");
return response;
}
}
from agents_framework import HumanInTheLoop
@function_tool
def delete_file(path: str) -> str:
"""Delete a file (requires approval)."""
return f"Deleted {path}"
# 添加审批包装器
delete_file_with_approval = HumanInTheLoop(
tool=delete_file,
approval_prompt="Approve deletion of {path}?"
)
agent = Agent(tools=[delete_file_with_approval])
workflow = GraphWorkflow()
# 添加多个智能体
workflow.add_node("analyst1", analyst_agent)
workflow.add_node("analyst2", analyst_agent)
workflow.add_node("synthesizer", synthesis_agent)
# 并行执行
workflow.add_edge("START", ["analyst1", "analyst2"]) # 两者并行运行
workflow.add_edge(["analyst1", "analyst2"], "synthesizer") # 等待两者完成
result = await workflow.run(message="Analyze market trends")
from pydantic import BaseModel
class WeatherReport(BaseModel):
location: str
temperature: float
conditions: str
agent = Agent(
model=model,
instructions="Generate weather reports",
response_format=WeatherReport
)
response = await agent.run(message="Weather in Seattle")
report: WeatherReport = response.parsed
print(f"{report.location}: {report.temperature}°F, {report.conditions}")
from agents_framework import RetryPolicy
agent = Agent(
model=model,
retry_policy=RetryPolicy(
max_retries=3,
backoff_factor=2.0,
exceptions=[TimeoutError, ConnectionError]
)
)
try:
response = await agent.run(message="Hello")
except Exception as e:
print(f"Failed after retries: {e}")
在以下情况下使用 Microsoft Agent Framework :
在以下情况下使用 amplihack 原生智能体 :
混合方法 :
# 使用 amplihack 进行编排
from claude import Agent as ClaudeAgent
orchestrator = ClaudeAgent("orchestrator.md")
# 将有状态智能体任务委派给 Agent Framework
from agents_framework import Agent, Thread
conversational_agent = Agent(
model=ModelClient(model="gpt-4"),
instructions="Maintain conversation context"
)
thread = Thread()
response1 = await conversational_agent.run(thread=thread, message="Start task")
response2 = await conversational_agent.run(thread=thread, message="Continue")
# 使用 amplihack 进行最终合成
result = orchestrator.process({"responses": [response1, response2]})
完整模式请参阅 @integration/amplihack-integration.md。
安装 : pip install agent-framework-core --pre (Python) 或 dotnet add package Microsoft.Agents.AI --prerelease (C#)
创建基础智能体 :
from agents_framework import Agent, ModelClient
agent = Agent(
name="assistant",
model=ModelClient(model="gpt-4"),
instructions="You are a helpful assistant"
)
response = await agent.run(message="Hello!")
添加工具 :
@function_tool
def calculate(expr: str) -> float:
return eval(expr)
agent = Agent(model=model, tools=[calculate])
构建工作流 :
workflow = GraphWorkflow()
workflow.add_node("agent1", agent1)
workflow.add_node("agent2", agent2)
workflow.add_edge("agent1", "agent2")
result = await workflow.run(message="Task")
添加遥测 :
from opentelemetry import trace
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("agent-run"):
response = await agent.run(message="Hello")
详细信息请参阅:
@reference/01-overview.md - 架构、组件、用例@reference/02-agents.md - 智能体创建、生命周期、高级特性@reference/03-workflows.md - 工作流模式、执行器、检查点@reference/04-tools-functions.md - 工具定义、审批工作流、错误处理@reference/05-context-middleware.md - 上下文提供者、中间件模式、身份验证@reference/06-telemetry-monitoring.md - OpenTelemetry、日志记录、调试@reference/07-advanced-patterns.md - 多智能体模式、流式传输、DevUI@examples/01-basic-agent.py - 简单对话智能体@examples/02-tool-integration.py - 具有函数调用功能的智能体@examples/03-simple-workflow.py - 多智能体工作流@examples/04-basic-agent.cs - C# 智能体实现@examples/05-tool-integration.cs - C# 工具集成@examples/06-simple-workflow.cs - C# 工作流示例检查框架新鲜度:python @scripts/check-freshness.py
当前版本跟踪:@metadata/version-tracking.json
令牌计数 : ~4,200 tokens (低于 4,800 限制)
每周安装数
143
仓库
GitHub Stars
39
首次出现
Jan 23, 2026
安全审计
安装于
opencode123
github-copilot123
codex117
gemini-cli116
cursor113
kimi-cli109
Version : 0.1.0-preview | Last Updated : 2025-11-15 | Framework Version : 0.1.0-preview Languages : Python 3.10+, C# (.NET 8.0+) | License : MIT
Microsoft Agent Framework is an open-source platform for building production AI agents and workflows, unifying AutoGen's simplicity with Semantic Kernel's enterprise features.
Core Capabilities : AI Agents (stateful conversations, tool integration) | Workflows (graph-based orchestration, parallel processing) | Enterprise features (telemetry, middleware, MCP support)
Installation :
pip install agent-framework-core --predotnet add package Microsoft.Agents.AI --prereleaseRepository : https://github.com/microsoft/agent-framework (5.1k stars)
Use Microsoft Agent Framework when you need:
Integration with amplihack : Use Agent Framework for stateful conversational agents and complex orchestration. Use amplihack's native agent system for stateless task delegation and simple orchestration. See @integration/decision-framework.md for detailed guidance.
Stateful conversational entities that process messages, call tools, and maintain context.
Python Example :
from agents_framework import Agent, ModelClient
# Create agent with model
agent = Agent(
name="assistant",
model=ModelClient(model="gpt-4"),
instructions="You are a helpful assistant"
)
# Single-turn conversation
response = await agent.run(message="Hello!")
print(response.content)
# Multi-turn with thread
from agents_framework import Thread
thread = Thread()
response = await agent.run(thread=thread, message="What's 2+2?")
response = await agent.run(thread=thread, message="Double that")
C# Example :
using Microsoft.Agents.AI;
var agent = new Agent(
name: "assistant",
model: new ModelClient(model: "gpt-4"),
instructions: "You are a helpful assistant"
);
var response = await agent.RunAsync("Hello!");
Console.WriteLine(response.Content);
Extend agent capabilities by providing callable functions.
Python Example :
from agents_framework import function_tool
@function_tool
def get_weather(location: str) -> str:
"""Get weather for a location."""
return f"Weather in {location}: Sunny, 72°F"
agent = Agent(
name="assistant",
model=ModelClient(model="gpt-4"),
tools=[get_weather]
)
response = await agent.run(message="What's the weather in Seattle?")
# Agent automatically calls get_weather() and responds with result
C# Example :
[FunctionTool]
public static string GetWeather(string location)
{
return $"Weather in {location}: Sunny, 72°F";
}
var agent = new Agent(
name: "assistant",
model: new ModelClient(model: "gpt-4"),
tools: new[] { typeof(Tools).GetMethod("GetWeather") }
);
Graph-based orchestration for multi-agent systems with conditional routing and parallel execution.
Python Example :
from agents_framework import Workflow, GraphWorkflow
# Define workflow graph
workflow = GraphWorkflow()
# Add agents as nodes
workflow.add_node("researcher", research_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("reviewer", review_agent)
# Define edges (control flow)
workflow.add_edge("researcher", "writer") # Sequential
workflow.add_edge("writer", "reviewer")
# Conditional routing
def should_revise(state):
return state.get("needs_revision", False)
workflow.add_conditional_edge(
"reviewer",
should_revise,
{"revise": "writer", "done": "END"}
)
# Execute workflow
result = await workflow.run(initial_message="Research AI trends")
C# Example :
var workflow = new GraphWorkflow();
workflow.AddNode("researcher", researchAgent);
workflow.AddNode("writer", writerAgent);
workflow.AddNode("reviewer", reviewAgent);
workflow.AddEdge("researcher", "writer");
workflow.AddEdge("writer", "reviewer");
var result = await workflow.RunAsync("Research AI trends");
Maintain conversation history and shared state across agents.
Python :
from agents_framework import Thread, ContextProvider
# Thread maintains conversation history
thread = Thread()
await agent.run(thread=thread, message="Remember: My name is Alice")
await agent.run(thread=thread, message="What's my name?") # "Alice"
# Custom context provider
class DatabaseContext(ContextProvider):
async def get_context(self, thread_id: str):
return await db.fetch_history(thread_id)
async def save_context(self, thread_id: str, messages):
await db.save_history(thread_id, messages)
agent = Agent(model=model, context_provider=DatabaseContext())
Add cross-cutting concerns like logging, auth, and monitoring.
Python :
from agents_framework import Middleware
from opentelemetry import trace
# Custom middleware
class LoggingMiddleware(Middleware):
async def process(self, message, next_handler):
print(f"Processing: {message.content}")
response = await next_handler(message)
print(f"Response: {response.content}")
return response
# OpenTelemetry integration
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("agent-run"):
response = await agent.run(message="Hello")
C# :
public class LoggingMiddleware : IMiddleware
{
public async Task<Message> ProcessAsync(Message message, Func<Message, Task<Message>> next)
{
Console.WriteLine($"Processing: {message.Content}");
var response = await next(message);
Console.WriteLine($"Response: {response.Content}");
return response;
}
}
from agents_framework import HumanInTheLoop
@function_tool
def delete_file(path: str) -> str:
"""Delete a file (requires approval)."""
return f"Deleted {path}"
# Add approval wrapper
delete_file_with_approval = HumanInTheLoop(
tool=delete_file,
approval_prompt="Approve deletion of {path}?"
)
agent = Agent(tools=[delete_file_with_approval])
workflow = GraphWorkflow()
# Add multiple agents
workflow.add_node("analyst1", analyst_agent)
workflow.add_node("analyst2", analyst_agent)
workflow.add_node("synthesizer", synthesis_agent)
# Parallel execution
workflow.add_edge("START", ["analyst1", "analyst2"]) # Both run in parallel
workflow.add_edge(["analyst1", "analyst2"], "synthesizer") # Wait for both
result = await workflow.run(message="Analyze market trends")
from pydantic import BaseModel
class WeatherReport(BaseModel):
location: str
temperature: float
conditions: str
agent = Agent(
model=model,
instructions="Generate weather reports",
response_format=WeatherReport
)
response = await agent.run(message="Weather in Seattle")
report: WeatherReport = response.parsed
print(f"{report.location}: {report.temperature}°F, {report.conditions}")
from agents_framework import RetryPolicy
agent = Agent(
model=model,
retry_policy=RetryPolicy(
max_retries=3,
backoff_factor=2.0,
exceptions=[TimeoutError, ConnectionError]
)
)
try:
response = await agent.run(message="Hello")
except Exception as e:
print(f"Failed after retries: {e}")
Use Microsoft Agent Framework when :
Use amplihack native agents when :
Hybrid Approach :
# Use amplihack for orchestration
from claude import Agent as ClaudeAgent
orchestrator = ClaudeAgent("orchestrator.md")
# Delegate to Agent Framework for stateful agents
from agents_framework import Agent, Thread
conversational_agent = Agent(
model=ModelClient(model="gpt-4"),
instructions="Maintain conversation context"
)
thread = Thread()
response1 = await conversational_agent.run(thread=thread, message="Start task")
response2 = await conversational_agent.run(thread=thread, message="Continue")
# Use amplihack for final synthesis
result = orchestrator.process({"responses": [response1, response2]})
See @integration/amplihack-integration.md for complete patterns.
Install : pip install agent-framework-core --pre (Python) or dotnet add package Microsoft.Agents.AI --prerelease (C#)
Create Basic Agent :
from agents_framework import Agent, ModelClient
agent = Agent(
name="assistant",
model=ModelClient(model="gpt-4"),
instructions="You are a helpful assistant"
)
response = await agent.run(message="Hello!")
Add Tools :
@function_tool
def calculate(expr: str) -> float:
return eval(expr)
agent = Agent(model=model, tools=[calculate])
Build Workflow :
workflow = GraphWorkflow()
workflow.add_node("agent1", agent1)
workflow.add_node("agent2", agent2)
workflow.add_edge("agent1", "agent2")
result = await workflow.run(message="Task")
For detailed information, see:
@reference/01-overview.md - Architecture, components, use cases@reference/02-agents.md - Agent creation, lifecycle, advanced features@reference/03-workflows.md - Workflow patterns, executors, checkpointing@reference/04-tools-functions.md - Tool definition, approval workflows, error handling@reference/05-context-middleware.md - Context providers, middleware patterns, auth@reference/06-telemetry-monitoring.md - OpenTelemetry, logging, debugging@reference/07-advanced-patterns.md - Multi-agent patterns, streaming, DevUI@examples/01-basic-agent.py - Simple conversational agent@examples/02-tool-integration.py - Agent with function calling@examples/03-simple-workflow.py - Multi-agent workflow@examples/04-basic-agent.cs - C# agent implementation@examples/05-tool-integration.cs - C# tool integration@examples/06-simple-workflow.cs - C# workflow exampleCheck framework freshness: python @scripts/check-freshness.py
Current version tracking: @metadata/version-tracking.json
Token Count : ~4,200 tokens (under 4,800 limit)
Weekly Installs
143
Repository
GitHub Stars
39
First Seen
Jan 23, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykWarn
Installed on
opencode123
github-copilot123
codex117
gemini-cli116
cursor113
kimi-cli109
AI Elements:基于shadcn/ui的AI原生应用组件库,快速构建对话界面
69,600 周安装
Expo React Native TypeScript 移动开发最佳实践指南 | 性能优化与代码规范
433 周安装
Flutter专家技能指南:精通Flutter 3.x多平台开发、性能优化与架构设计
439 周安装
Bash脚本编写工作流:生产级自动化脚本开发与防御性编程指南
433 周安装
Vercel一键部署工具 - 无需认证,自动检测框架,支持React/Vue/Svelte等主流框架
445 周安装
价值股息筛选器:两阶段方法识别高质量股息股票,优化API使用效率
457 周安装
Angular Core 核心架构深度解析:Ivy渲染、依赖注入与变更检测机制
481 周安装
Add Telemetry :
from opentelemetry import trace
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("agent-run"):
response = await agent.run(message="Hello")