Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
961 changes: 75 additions & 886 deletions docs/changelog.md

Large diffs are not rendered by default.

171 changes: 171 additions & 0 deletions docs/tutorial/adding-agent-with-tools.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,171 @@
---
sidebar_position: 4
---

**Tutorial Progress: Step 3 of 5**
[← Previous: Your First RAG Pipeline](./first-rag-pipeline) | [Next: Building a Graph Workflow →](./building-graph-workflow)

# Step 3: Adding an Agent with Tools

In this step you will turn your RAG pipeline into an intelligent agent that can use tools, make decisions, and call external functions.

## Why agents?

RAG is great for question answering. Agents add reasoning, tool use, and multi-step planning on top of retrieval.

## 1. The FunctionCallingAgent

SynapseKit's `FunctionCallingAgent` is a typed, async-first agent that supports multiple LLM providers and automatic tool schema generation.

```python
from synapsekit import FunctionCallingAgent
from synapsekit.tools import Tool

agent = FunctionCallingAgent(
model="gpt-4o",
tools=[...], # list of Tool objects
memory=..., # optional
guardrails=..., # optional
)
```

## 2. Creating a custom tool

The easiest way is using the `@tool` decorator or the `Tool` class.

```python
from synapsekit.tools import tool
from pydantic import BaseModel

class WebSearchInput(BaseModel):
query: str
max_results: int = 5

@tool("web_search", "Search the web for current information")
async def web_search(input: WebSearchInput) -> list[str]:
# In real code you would call Tavily, SerpAPI, etc.
return [
f"Result 1 for {input.query}",
f"Result 2 for {input.query}",
]
```

You can also wrap existing functions:

```python
def calculate(expression: str) -> float:
return eval(expression) # demo only

calc_tool = Tool.from_function(
calculate,
name="calculator",
description="Evaluate a math expression"
)
```

## 3. Full agent example with RAG + tools

```python
import asyncio
from synapsekit import FunctionCallingAgent, RAGPipeline
from synapsekit.tools import tool
from pydantic import BaseModel

class SearchInput(BaseModel):
query: str

@tool("company_search", "Search internal company knowledge base")
async def company_search(input: SearchInput):
rag = RAGPipeline(model="gpt-4o-mini", vector_store="chroma")
rag.add_directory("company_docs/")
return rag.ask_sync(input.query)

agent = FunctionCallingAgent(
model="gpt-4o",
tools=[company_search],
system_prompt="You are a helpful company assistant. Use tools when needed.",
)

async def main():
result = await agent.run("What is our Q3 revenue target?")
print(result)

asyncio.run(main())
```

## 4. Adding guardrails and budget control

Production agents need safety.

```python
from synapsekit.guardrails import ContentFilter, BudgetGuard

agent = FunctionCallingAgent(
model="gpt-4o",
tools=[...],
guardrails=[
ContentFilter(block_categories=["hate", "violence"]),
BudgetGuard(max_tokens_per_turn=4000, max_cost_usd=0.05),
],
)
```

## 5. Streaming agent steps (advanced)

Agents can stream intermediate reasoning steps.

```python
async for event in agent.stream_events("Research SynapseKit competitors"):
if event.type == "tool_call":
print(f"Calling tool: {event.tool_name}")
elif event.type == "final_answer":
print(event.content)
```

## 6. Complete runnable agent tutorial code

```python
import asyncio
from synapsekit import FunctionCallingAgent
from synapsekit.tools import tool
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
city: str = Field(..., description="City name")

@tool("get_weather", "Get current weather for a city")
async def get_weather(input: WeatherInput) -> str:
# Mock implementation
return f"The weather in {input.city} is sunny, 24°C"

agent = FunctionCallingAgent(
model="gpt-4o-mini",
tools=[get_weather],
system_prompt="You are a friendly weather assistant.",
)

async def chat():
print("Weather Agent ready. Type 'exit' to quit.\n")
while True:
user_input = input("You: ")
if user_input.lower() == "exit":
break
response = await agent.run(user_input)
print(f"Agent: {response}\n")

if __name__ == "__main__":
asyncio.run(chat())
```

## Key concepts covered

- Tool schema generation from Pydantic models
- Automatic function calling loop
- Guardrails & budget protection
- Streaming intermediate steps
- Memory integration (agent remembers previous turns)

**You now have a capable agent!**

**Tutorial Progress: Step 3 of 5**
[← Previous: Your First RAG Pipeline](./first-rag-pipeline) | [Next: Building a Graph Workflow →](./building-graph-workflow)
167 changes: 167 additions & 0 deletions docs/tutorial/building-graph-workflow.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,167 @@
---
sidebar_position: 5
---

**Tutorial Progress: Step 4 of 5**
[← Previous: Adding an Agent with Tools](./adding-agent-with-tools) | [Next: Deploying to Production →](./deploying-to-production)

# Step 4: Building a Graph Workflow

In this step you will compose RAG and Agents into a stateful, resumable graph workflow using `StateGraph`.

## Why graphs?

Graphs give you explicit control over control flow, cycles, human-in-the-loop, checkpointing, and visualization — things that become messy with pure agent loops.

## 1. Core concepts

- **State**: A typed dictionary that flows through the graph
- **Nodes**: Functions or agents that transform state
- **Edges**: Connections (normal, conditional, cycles)
- **Checkpointing**: Automatic persistence of state after each step

## 2. A simple linear graph

```python
from synapsekit import StateGraph
from typing import TypedDict

class WorkflowState(TypedDict):
question: str
answer: str
sources: list[str]

async def retrieve(state: WorkflowState):
# Call your RAG pipeline here
return {"sources": ["doc1.pdf", "doc2.md"]}

async def generate(state: WorkflowState):
# Call LLM with retrieved context
return {"answer": "SynapseKit is awesome because..."}

graph = StateGraph(WorkflowState)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_edge("retrieve", "generate")
graph.set_entry_point("retrieve")

app = graph.compile()
```

## 3. Adding conditional routing

```python
def route_after_retrieval(state: WorkflowState):
if len(state["sources"]) > 3:
return "generate_detailed"
return "generate_concise"

graph.add_conditional_edges(
"retrieve",
route_after_retrieval,
{
"generate_detailed": "generate_detailed",
"generate_concise": "generate_concise",
}
)
```

## 4. Human-in-the-loop (HITL)

```python
from synapsekit.graph import interrupt

async def human_review(state):
# Pause execution and wait for human input
return interrupt("Please review the draft answer", value=state["answer"])
```

The graph will stop and return control to your application.

## 5. Checkpointing & resumability

```python
from synapsekit.checkpointing import SqliteCheckpointer

checkpointer = SqliteCheckpointer(db_path="workflows.db")

app = graph.compile(checkpointer=checkpointer)

# Run with thread_id for resumability
config = {"configurable": {"thread_id": "user-123"}}

result = await app.ainvoke({"question": "Explain RAG"}, config=config)
```

Later you can resume from any checkpoint.

## 6. Full example: Research → Draft → Review → Publish graph

```python
import asyncio
from synapsekit import StateGraph
from typing import TypedDict, Literal

class ResearchState(TypedDict):
topic: str
research: str
draft: str
feedback: str | None
final: str | None

async def research_node(state):
# Call agent or RAG
return {"research": f"Research results about {state['topic']}"}

async def draft_node(state):
return {"draft": f"Draft article about {state['topic']}"}

async def review_node(state):
# Simulate human review
return {"feedback": "Looks good, minor changes needed"}

async def publish_node(state):
return {"final": state["draft"] + "\n\n[Published]"}

def should_revise(state):
if state.get("feedback"):
return "revise"
return "publish"

graph = StateGraph(ResearchState)
graph.add_node("research", research_node)
graph.add_node("draft", draft_node)
graph.add_node("review", review_node)
graph.add_node("publish", publish_node)

graph.add_edge("research", "draft")
graph.add_edge("draft", "review")
graph.add_conditional_edges("review", should_revise, {
"revise": "draft",
"publish": "publish"
})
graph.set_entry_point("research")

app = graph.compile()

async def main():
result = await app.ainvoke({"topic": "SynapseKit graphs"})
print(result["final"])

asyncio.run(main())
```

## 7. Visualization

```python
from synapsekit.graph import draw_mermaid

print(draw_mermaid(app))
```

Generates a Mermaid diagram you can render in docs or notebooks.

**You have now built a real multi-step workflow with branching and checkpointing!**

**Tutorial Progress: Step 4 of 5**
[← Previous: Adding an Agent with Tools](./adding-agent-with-tools) | [Next: Deploying to Production →](./deploying-to-production)
Loading