- LangGraph - Crie fluxos de trabalho de Agentes complexos e com várias etapas
- OpenAI Agents SDK - Framework oficial de Agentes da OpenAI
- Microsoft AutoGen - Framework de conversação multiagente da Microsoft
- Google ADK - Google Agent Development Kit
Padrão Principal de Integração
Todas as integrações de frameworks seguem este padrão principal:from novita_sandbox.agent_runtime import AgentRuntimeApp
# 1. Create Agent Runtime application instance
app = AgentRuntimeApp()
# 2. Initialize your Agent framework
# 3. Define entry point with decorator
@app.entrypoint
def agent_invocation(request: dict) -> dict:
"""
Args:
request: Request data, typically contains fields like prompt
Returns:
Response data dictionary
"""
prompt = request.get("prompt", "")
# Call your Agent framework
result = your_agent.run(prompt)
return {"result": result}
# 4. Run the application
if __name__ == "__main__":
app.run()
LangGraph
LangGraph é o framework oficial da LangChain para criar aplicações de Agentes com estado e várias etapas.Código de Exemplo
Para ver o projeto de exemplo completo, consulte aquifrom langchain_community.chat_models import ChatOpenAI
from typing_extensions import TypedDict
from typing import Annotated
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
# Import Novita Agent Runtime
from novita_sandbox.agent_runtime import AgentRuntimeApp
app = AgentRuntimeApp()
# Define state
class State(TypedDict):
messages: Annotated[list, add_messages]
# Initialize LLM
llm = ChatOpenAI(model="gpt-4")
# Define tools
def get_weather(location: str) -> str:
"""Get weather information for a specified location"""
return f"The weather in {location} is sunny, 23°C"
tools = [get_weather]
llm_with_tools = llm.bind_tools(tools)
# Define node function
def chatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
# Build graph
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools=tools))
graph_builder.add_conditional_edges(
"chatbot",
tools_condition,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()
# Define entry point
@app.entrypoint
def agent_invocation(request: dict) -> dict:
"""LangGraph Agent entry point"""
prompt = request.get("prompt", "Hello!")
# Invoke LangGraph
result = graph.invoke({
"messages": [{"role": "user", "content": prompt}]
})
# Extract the last message
final_message = result['messages'][-1].content
return {"result": final_message}
if __name__ == "__main__":
app.run()
OpenAI Agents SDK
OpenAI Agents SDK é o kit de ferramentas oficial da OpenAI para criar Agentes de IA.Código de Exemplo
Para ver o projeto de exemplo completo, consulte aquiimport os
from novita_sandbox.agent_runtime import AgentRuntimeApp
from openai import AsyncOpenAI
# Create application
app = AgentRuntimeApp()
# Define tool function
def get_weather(city: str) -> str:
"""Get city weather (simulated)"""
weather_data = {
"Beijing": "Sunny, 15°C",
"Shanghai": "Cloudy, 20°C",
"Shenzhen": "Light rain, 25°C",
}
return weather_data.get(city, f"{city}: Sunny, 23°C")
# Tool definition
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather information for a specified city",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City name, e.g. 'Beijing', 'Shanghai'"
}
},
"required": ["city"]
}
}
}
]
# Agent core logic
async def run_agent(query: str) -> str:
"""Run OpenAI Agent (supports function calling)"""
client = AsyncOpenAI(
base_url=os.getenv("OPENAI_API_BASE"),
api_key=os.getenv("NOVITA_API_KEY"),
)
messages = [
{"role": "system", "content": "You are a helpful AI assistant that can query weather."},
{"role": "user", "content": query}
]
# First call
response = await client.chat.completions.create(
model=os.getenv("MODEL_NAME", "deepseek/deepseek-v3.1-terminus"),
messages=messages,
tools=TOOLS,
tool_choice="auto"
)
response_message = response.choices[0].message
# If tool call is needed
if response_message.tool_calls:
messages.append(response_message)
for tool_call in response_message.tool_calls:
function_args = eval(tool_call.function.arguments)
function_response = get_weather(**function_args)
messages.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": "get_weather",
"content": function_response
})
# Second call to get final response
final_response = await client.chat.completions.create(
model=os.getenv("MODEL_NAME", "deepseek/deepseek-v3.1-terminus"),
messages=messages
)
return final_response.choices[0].message.content
return response_message.content
# Novita Agent Runtime entry point
@app.entrypoint
async def agent_invocation(request: dict) -> dict:
"""Entry point function"""
prompt = request.get("prompt", "Hello!")
result = await run_agent(prompt)
return {"result": result}
# Start application
if __name__ == "__main__":
app.run()
Microsoft AutoGen
AutoGen é um framework de conversação multiagente desenvolvido pela Microsoft que oferece suporte à colaboração e a conversas entre múltiplos agentes, capaz de trabalhar de forma independente ou com humanos.Código de Exemplo
Para ver o projeto de exemplo completo, consulte aquiimport os
from novita_sandbox.agent_runtime import AgentRuntimeApp
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ModelFamily, ModelInfo
# Create application
app = AgentRuntimeApp()
# Define tool function
async def get_weather(city: str) -> str:
"""Get weather for a specified city"""
weather_data = {
"Beijing": "Sunny, 15°C",
"Shanghai": "Cloudy, 20°C",
"Shenzhen": "Light rain, 25°C",
}
return weather_data.get(city, f"{city}: Sunny, 23°C")
# Create AutoGen Agent
def create_agent():
"""Create AutoGen Agent"""
model_client = OpenAIChatCompletionClient(
base_url=os.getenv("OPENAI_BASE_URL", "https://api.novita.ai/v3/openai"),
model=os.getenv("MODEL_NAME", "deepseek/deepseek-v3.1-terminus"),
api_key=os.getenv("OPENAI_API_KEY"),
model_info=ModelInfo(
vision=False,
function_calling=True,
json_output=True,
family=ModelFamily.UNKNOWN,
),
)
agent = AssistantAgent(
name="assistant",
model_client=model_client,
tools=[get_weather],
system_message="You are a helpful AI assistant that can query weather information.",
reflect_on_tool_use=True,
)
return agent
# Run Agent
async def run_agent(prompt: str) -> str:
"""Run AutoGen Agent"""
agent = create_agent()
# Create user message
message = TextMessage(content=prompt, source="user")
# Run Agent
response_message = await agent.on_messages([message])
# Extract response content
if response_message and hasattr(response_message, 'chat_message'):
return response_message.chat_message.content
return str(response_message)
# Novita Agent Runtime entry point
@app.entrypoint
async def agent_invocation(request: dict):
"""Entry point function"""
prompt = request.get("prompt", "Hello!")
result = await run_agent(prompt)
return {"result": result}
# Start application
if __name__ == "__main__":
app.run()
Google ADK
Google Agent Development Kit (ADK) é o kit de ferramentas do Google para desenvolvimento de Agentes.Exemplo Completo
Para ver o projeto de exemplo completo, consulte aquiimport os
import uuid
import asyncio
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.genai import types
from novita_sandbox.agent_runtime import AgentRuntimeApp
# Create application
app = AgentRuntimeApp()
APP_NAME = "google_search_agent"
# Create Google ADK Agent
root_agent = LlmAgent(
model=os.getenv("GEMINI_MODEL", "gemini-2.0-flash"),
name=APP_NAME,
instruction="I can answer your questions by searching the internet. Ask me anything!",
tools=[google_search]
)
# Create Session service and Runner
session_service = InMemorySessionService()
runner = Runner(
agent=root_agent,
app_name=APP_NAME,
session_service=session_service
)
# Run Agent
async def run_agent(query: str) -> str:
"""Run Google ADK Agent"""
user_id = "user_default"
session_id = str(uuid.uuid4())
# Create Session
await session_service.create_session(
app_name=APP_NAME,
user_id=user_id,
session_id=session_id
)
# Create user message
user_content = types.Content(
role='user',
parts=[types.Part(text=query)]
)
# Run Agent
result = ""
async for event in runner.run_async(
user_id=user_id,
session_id=session_id,
new_message=user_content
):
if event.is_final_response() and event.content and event.content.parts:
result = event.content.parts[0].text
return result
# Novita Agent Runtime entry point
@app.entrypoint
def agent_invocation(request: dict):
"""Entry point function"""
prompt = request.get("prompt", "Hello!")
result = asyncio.run(run_agent(prompt))
return {"result": result}
# Start application
if __name__ == "__main__":
app.run()